Extend the network client

Makefile

@@ -286,7 +286,7 @@ network-module-test: assets
 INTEGRATION_TEST_IN_DOCKER = docker exec core_test
 network-integration-test:
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml up -d
-	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v --ignore-glob="$(TESTDIR)/network/*ocrd_all*.py"
+	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v --ignore-glob="tests/network/*ocrd_all*.py"
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml down --remove-orphans
 
 network-integration-test-cicd:

src/ocrd/cli/__init__.py

@@ -35,6 +35,10 @@
 \b
 {config.describe('OCRD_MAX_PROCESSOR_CACHE')}
 \b
+{config.describe('OCRD_NETWORK_CLIENT_POLLING_SLEEP')}
+\b
+{config.describe('OCRD_NETWORK_CLIENT_POLLING_TIMEOUT')}
+\b
 {config.describe('OCRD_NETWORK_SERVER_ADDR_PROCESSING')}
 \b
 {config.describe('OCRD_NETWORK_SERVER_ADDR_WORKFLOW')}

src/ocrd_network/cli/client.py

@@ -1,9 +1,9 @@
 import click
+from json import dumps, loads
 from typing import Optional
-
 from ocrd.decorators import parameter_option
-from ocrd_network import Client
 from ocrd_utils import DEFAULT_METS_BASENAME
+from ..client import Client
 
 
 @click.group('client')
@@ -31,9 +31,24 @@ def processing_cli():
     pass
 
 
+@processing_cli.command('check-status')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-j', '--processing-job-id', required=True)
+def check_processing_job_status(address: Optional[str], processing_job_id: str):
+    """
+    Check the status of a previously submitted processing job.
+    """
+    client = Client(server_addr_processing=address)
+    job_status = client.check_job_status(processing_job_id)
+    assert job_status
+    print(f"Processing job status: {job_status}")
+
+
 @processing_cli.command('processor')
 @click.argument('processor_name', required=True, type=click.STRING)
-@click.option('--address')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
 @click.option('-m', '--mets', required=True, default=DEFAULT_METS_BASENAME)
 @click.option('-I', '--input-file-grp', default='OCR-D-INPUT')
 @click.option('-O', '--output-file-grp', default='OCR-D-OUTPUT')
@@ -42,26 +57,32 @@ def processing_cli():
 @click.option('--result-queue-name')
 @click.option('--callback-url')
 @click.option('--agent-type', default='worker')
-def send_processing_request(
-        address: Optional[str],
-        processor_name: str,
-        mets: str,
-        input_file_grp: str,
-        output_file_grp: Optional[str],
-        page_id: Optional[str],
-        parameter: Optional[dict],
-        result_queue_name: Optional[str],
-        callback_url: Optional[str],
-        # TODO: This is temporally available to toggle
-        #  between the ProcessingWorker/ProcessorServer
-        agent_type: Optional[str]
+@click.option('-b', '--block', default=False,
+              help='If set, the client will block till job timeout, fail or success.')
+def send_processing_job_request(
+    address: Optional[str],
+    processor_name: str,
+    mets: str,
+    input_file_grp: str,
+    output_file_grp: Optional[str],
+    page_id: Optional[str],
+    parameter: Optional[dict],
+    result_queue_name: Optional[str],
+    callback_url: Optional[str],
+    # TODO: This is temporally available to toggle
+    #  between the ProcessingWorker/ProcessorServer
+    agent_type: Optional[str],
+    block: Optional[bool]
 ):
+    """
+    Submit a processing job to the processing server.
+    """
     req_params = {
         "path_to_mets": mets,
         "description": "OCR-D Network client request",
         "input_file_grps": input_file_grp.split(','),
         "parameters": parameter if parameter else {},
-        "agent_type": agent_type,
+        "agent_type": agent_type
     }
     if output_file_grp:
         req_params["output_file_grps"] = output_file_grp.split(',')
@@ -71,16 +92,13 @@ def send_processing_request(
         req_params["result_queue_name"] = result_queue_name
     if callback_url:
         req_params["callback_url"] = callback_url
-
-    client = Client(
-        server_addr_processing=address
-    )
-    response = client.send_processing_request(
-        processor_name=processor_name,
-        req_params=req_params
-    )
-    processing_job_id = response.get('job_id', None)
+    client = Client(server_addr_processing=address)
+    processing_job_id = client.send_processing_job_request(
+        processor_name=processor_name, req_params=loads(dumps(req_params)))
+    assert processing_job_id
     print(f"Processing job id: {processing_job_id}")
+    if block:
+        client.poll_job_status(job_id=processing_job_id)
 
 
 @client_cli.group('workflow')
@@ -91,6 +109,44 @@ def workflow_cli():
     pass
 
 
+@workflow_cli.command('check-status')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-j', '--workflow-job-id', required=True)
+def check_workflow_job_status(address: Optional[str], workflow_job_id: str):
+    """
+    Check the status of a previously submitted workflow job.
+    """
+    client = Client(server_addr_processing=address)
+    job_status = client.check_workflow_status(workflow_job_id)
+    assert job_status
+    print(f"Workflow job status: {job_status}")
+
+
+@workflow_cli.command('run')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-m', '--path-to-mets', required=True)
+@click.option('-w', '--path-to-workflow', required=True)
+@click.option('-b', '--block', default=False,
+              help='If set, the client will block till job timeout, fail or success.')
+def send_workflow_job_request(
+    address: Optional[str],
+    path_to_mets: str,
+    path_to_workflow: str,
+    block: Optional[bool]
+):
+    """
+    Submit a workflow job to the processing server.
+    """
+    client = Client(server_addr_processing=address)
+    workflow_job_id = client.send_workflow_job_request(path_to_wf=path_to_workflow, path_to_mets=path_to_mets)
+    assert workflow_job_id
+    print(f"Workflow job id: {workflow_job_id}")
+    if block:
+        client.poll_workflow_status(job_id=workflow_job_id)
+
+
 @client_cli.group('workspace')
 def workspace_cli():
     """

src/ocrd_network/client.py

@@ -1,37 +1,49 @@
-from json import dumps, loads
-from requests import post as requests_post
 from ocrd_utils import config, getLogger, LOG_FORMAT
+from .client_utils import (
+    get_ps_processing_job_status,
+    get_ps_workflow_job_status,
+    poll_job_status_till_timeout_fail_or_success,
+    poll_wf_status_till_timeout_fail_or_success,
+    post_ps_processing_request,
+    post_ps_workflow_request,
+    verify_server_protocol
+)
 
-from .constants import NETWORK_PROTOCOLS
 
-
-# TODO: This is just a conceptual implementation and first try to
-#  trigger further discussions on how this should look like.
 class Client:
     def __init__(
         self,
-        server_addr_processing: str = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING,
-        server_addr_workflow: str = config.OCRD_NETWORK_SERVER_ADDR_WORKFLOW,
-        server_addr_workspace: str = config.OCRD_NETWORK_SERVER_ADDR_WORKSPACE
+        server_addr_processing: str,
+        timeout: int = config.OCRD_NETWORK_CLIENT_POLLING_TIMEOUT,
+        wait: int = config.OCRD_NETWORK_CLIENT_POLLING_SLEEP
     ):
         self.log = getLogger(f"ocrd_network.client")
+        if not server_addr_processing:
+            server_addr_processing = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING
         self.server_addr_processing = server_addr_processing
-        self.server_addr_workflow = server_addr_workflow
-        self.server_addr_workspace = server_addr_workspace
-
-    def send_processing_request(self, processor_name: str, req_params: dict):
         verify_server_protocol(self.server_addr_processing)
-        req_url = f"{self.server_addr_processing}/processor/{processor_name}"
-        req_headers = {"Content-Type": "application/json; charset=utf-8"}
-        req_json = loads(dumps(req_params))
-        self.log.info(f"Sending processing request to: {req_url}")
-        self.log.debug(req_json)
-        response = requests_post(url=req_url, headers=req_headers, json=req_json)
-        return response.json()
-
-
-def verify_server_protocol(address: str):
-    for protocol in NETWORK_PROTOCOLS:
-        if address.startswith(protocol):
-            return
-    raise ValueError(f"Wrong/Missing protocol in the server address: {address}, must be one of: {NETWORK_PROTOCOLS}")
+        self.polling_timeout = timeout
+        self.polling_wait = wait
+        self.polling_tries = int(timeout/wait)
+
+    def check_job_status(self, job_id: str):
+        return get_ps_processing_job_status(self.server_addr_processing, processing_job_id=job_id)
+
+    def check_workflow_status(self, workflow_job_id: str):
+        return get_ps_workflow_job_status(self.server_addr_processing, workflow_job_id=workflow_job_id)
+
+    def poll_job_status(self, job_id: str) -> str:
+        return poll_job_status_till_timeout_fail_or_success(
+            ps_server_host=self.server_addr_processing, job_id=job_id, tries=self.polling_tries, wait=self.polling_wait)
+
+    def poll_workflow_status(self, job_id: str) -> str:
+        return poll_wf_status_till_timeout_fail_or_success(
+            ps_server_host=self.server_addr_processing, job_id=job_id, tries=self.polling_tries, wait=self.polling_wait)
+
+    def send_processing_job_request(self, processor_name: str, req_params: dict) -> str:
+        return post_ps_processing_request(
+            ps_server_host=self.server_addr_processing, processor=processor_name, job_input=req_params)
+
+    def send_workflow_job_request(self, path_to_wf: str, path_to_mets: str):
+        return post_ps_workflow_request(
+            ps_server_host=self.server_addr_processing, path_to_wf=path_to_wf, path_to_mets=path_to_mets)

src/ocrd_network/client_utils.py

@@ -0,0 +1,81 @@
+from requests import get as request_get, post as request_post
+from time import sleep
+from .constants import JobState, NETWORK_PROTOCOLS
+
+
+def _poll_endpoint_status(ps_server_host: str, job_id: str, job_type: str, tries: int, wait: int):
+    if job_type not in ["workflow", "processor"]:
+        raise ValueError(f"Unknown job type '{job_type}', expected 'workflow' or 'processor'")
+    job_state = JobState.unset
+    while tries > 0:
+        sleep(wait)
+        if job_type == "processor":
+            job_state = get_ps_processing_job_status(ps_server_host, job_id)
+        if job_type == "workflow":
+            job_state = get_ps_workflow_job_status(ps_server_host, job_id)
+        if job_state == JobState.success or job_state == JobState.failed:
+            break
+        tries -= 1
+    return job_state
+
+
+def poll_job_status_till_timeout_fail_or_success(ps_server_host: str, job_id: str, tries: int, wait: int) -> JobState:
+    return _poll_endpoint_status(ps_server_host, job_id, "processor", tries, wait)
+
+
+def poll_wf_status_till_timeout_fail_or_success(ps_server_host: str, job_id: str, tries: int, wait: int) -> JobState:
+    return _poll_endpoint_status(ps_server_host, job_id, "workflow", tries, wait)
+
+
+def get_ps_processing_job_status(ps_server_host: str, processing_job_id: str) -> str:
+    request_url = f"{ps_server_host}/processor/job/{processing_job_id}"
+    response = request_get(url=request_url, headers={"accept": "application/json; charset=utf-8"})
+    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    job_state = response.json()["state"]
+    assert job_state
+    return job_state
+
+
+def get_ps_workflow_job_status(ps_server_host: str, workflow_job_id: str) -> str:
+    request_url = f"{ps_server_host}/workflow/job-simple/{workflow_job_id}"
+    response = request_get(url=request_url, headers={"accept": "application/json; charset=utf-8"})
+    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    job_state = response.json()["state"]
+    assert job_state
+    return job_state
+
+
+def post_ps_processing_request(ps_server_host: str, processor: str, job_input: dict) -> str:
+    request_url = f"{ps_server_host}/processor/run/{processor}"
+    response = request_post(
+        url=request_url,
+        headers={"accept": "application/json; charset=utf-8"},
+        json=job_input
+    )
+    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    processing_job_id = response.json()["job_id"]
+    assert processing_job_id
+    return processing_job_id
+
+
+# TODO: Can be extended to include other parameters such as page_wise
+def post_ps_workflow_request(ps_server_host: str, path_to_wf: str, path_to_mets: str) -> str:
+    request_url = f"{ps_server_host}/workflow/run?mets_path={path_to_mets}&page_wise=True"
+    response = request_post(
+        url=request_url,
+        headers={"accept": "application/json; charset=utf-8"},
+        files={"workflow": open(path_to_wf, "rb")}
+    )
+    # print(response.json())
+    # print(response.__dict__)
+    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    wf_job_id = response.json()["job_id"]
+    assert wf_job_id
+    return wf_job_id
+
+
+def verify_server_protocol(address: str):
+    for protocol in NETWORK_PROTOCOLS:
+        if address.startswith(protocol):
+            return
+    raise ValueError(f"Wrong/Missing protocol in the server address: {address}, must be one of: {NETWORK_PROTOCOLS}")

src/ocrd_network/server_utils.py

@@ -193,7 +193,9 @@ def parse_workflow_tasks(logger: Logger, workflow_content: str) -> List[Processo
 
 
 def raise_http_exception(logger: Logger, status_code: int, message: str, error: Exception = None) -> None:
-    logger.exception(f"{message} {error}")
+    if error:
+        message = f"{message} {error}"
+    logger.exception(f"{message}")
     raise HTTPException(status_code=status_code, detail=message)
 
 
@@ -210,12 +212,12 @@ def validate_job_input(logger: Logger, processor_name: str, ocrd_tool: dict, job
         raise_http_exception(logger, status.HTTP_404_NOT_FOUND, message)
     try:
         report = ParameterValidator(ocrd_tool).validate(dict(job_input.parameters))
-        if not report.is_valid:
-            message = f"Failed to validate processing job input against the tool json of processor: {processor_name}\n"
-            raise_http_exception(logger, status.HTTP_404_BAD_REQUEST, message + report.errors)
     except Exception as error:
         message = f"Failed to validate processing job input against the ocrd tool json of processor: {processor_name}"
-        raise_http_exception(logger, status.HTTP_404_BAD_REQUEST, message, error)
+        raise_http_exception(logger, status.HTTP_400_BAD_REQUEST, message, error)
+    if report and not report.is_valid:
+        message = f"Failed to validate processing job input against the tool json of processor: {processor_name}\n"
+        raise_http_exception(logger, status.HTTP_400_BAD_REQUEST, f"{message}{report.errors}")
 
 
 def validate_workflow(logger: Logger, workflow: str) -> None:

src/ocrd_utils/config.py

@@ -145,6 +145,16 @@ def _ocrd_download_timeout_parser(val):
         description="Default address of Processing Server to connect to (for `ocrd network client processing`).",
         default=(True, ''))
 
+config.add("OCRD_NETWORK_CLIENT_POLLING_SLEEP",
+           description="How many seconds to sleep before trying again.",
+           parser=int,
+           default=(True, 30))
+
+config.add("OCRD_NETWORK_CLIENT_POLLING_TIMEOUT",
+           description="Timeout for a blocking ocrd network client (in seconds).",
+           parser=int,
+           default=(True, 3600))
+
 config.add("OCRD_NETWORK_SERVER_ADDR_WORKFLOW",
         description="Default address of Workflow Server to connect to (for `ocrd network client workflow`).",
         default=(True, ''))

tests/network/config.py

@@ -55,6 +55,20 @@
     parser=_ocrd_download_timeout_parser
 )
 
+test_config.add(
+    "OCRD_NETWORK_CLIENT_POLLING_SLEEP",
+    description="How many seconds to sleep before trying again.",
+    parser=int,
+    default=(True, 30)
+)
+
+test_config.add(
+    "OCRD_NETWORK_CLIENT_POLLING_TIMEOUT",
+    description="Timeout for a blocking ocrd network client (in seconds).",
+    parser=int,
+    default=(True, 3600)
+)
+
 test_config.add(
     name="OCRD_NETWORK_SERVER_ADDR_PROCESSING",
     description="Default address of Processing Server to connect to (for `ocrd network client processing`).",