Improve ocrd-network-client error handling

src/ocrd_network/cli/client.py

@@ -1,3 +1,4 @@
+import sys
 import click
 from json import dumps
 from typing import List, Optional, Tuple
@@ -10,6 +11,7 @@
 from ocrd_utils.introspect import set_json_key_value_overrides
 from ocrd_utils.str import parse_json_string_or_file
 from ..client import Client
+from ..client_utils import OcrdNetworkClientError
 
 
 ADDRESS_HELP = 'The URL of the Processing Server. If not provided, ' + \
@@ -53,8 +55,12 @@
     Each processor is shown only once regardless of the amount of deployed instances.
     """
     client = Client(server_addr_processing=address)
-    processors_list = client.check_deployed_processors()
-    print(dumps(processors_list, indent=4))
+    try:
+        processors_list = client.check_deployed_processors()
+        print(dumps(processors_list, indent=4))
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @discovery_cli.command('processor')
@@ -65,8 +71,12 @@
     Get the json tool of a deployed processor specified with `processor_name`
     """
     client = Client(server_addr_processing=address)
-    ocrd_tool = client.check_deployed_processor_ocrd_tool(processor_name=processor_name)
-    print(dumps(ocrd_tool, indent=4))
+    try:
+        ocrd_tool = client.check_deployed_processor_ocrd_tool(processor_name=processor_name)
+        print(dumps(ocrd_tool, indent=4))
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @client_cli.group('processing')
@@ -85,8 +95,12 @@
     Check the log of a previously submitted processing job.
     """
     client = Client(server_addr_processing=address)
-    response = client.check_job_log(job_id=processing_job_id)
-    print(response._content.decode(encoding='utf-8'))
+    try:
+        response = client.check_job_log(job_id=processing_job_id)
+        print(response._content.decode(encoding='utf-8'))
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @processing_cli.command('check-status')
@@ -97,9 +111,13 @@
     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}")
+    try:
+        job_status = client.check_job_status(processing_job_id)
+        assert job_status
+        print(f"Processing job status: {job_status}")
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @processing_cli.command('run')
@@ -154,12 +172,16 @@
     if callback_url:
         req_params["callback_url"] = callback_url
     client = Client(server_addr_processing=address)
-    processing_job_id = client.send_processing_job_request(
-        processor_name=processor_name, req_params=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, print_state=print_state)
+    try:
+        processing_job_id = client.send_processing_job_request(
+            processor_name=processor_name, req_params=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, print_state=print_state)
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @client_cli.group('workflow')
@@ -178,9 +200,13 @@
     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}")
+    try:
+        job_status = client.check_workflow_status(workflow_job_id)
+        assert job_status
+        print(f"Workflow job status: {job_status}")
+    except OcrdNetworkClientError as e:
+        print(f"{e.message}")
+        sys.exit(1)
 
 
 @workflow_cli.command('run')
@@ -225,7 +251,11 @@
     print(f"Workflow job id: {workflow_job_id}")
     if block:
         print(f"Polling state of workflow job {workflow_job_id}")
-        state = client.poll_workflow_status(job_id=workflow_job_id, print_state=print_state)
+        try:
+            state = client.poll_workflow_status(job_id=workflow_job_id, print_state=print_state)
+        except OcrdNetworkError as e:
+            print(f"{e.message}")
+            sys.exit(1)
         if state != JobState.success:
             print(f"Workflow failed with {state}")
             exit(1)

src/ocrd_network/client_utils.py

@@ -1,10 +1,23 @@
 import json
 import os
-from requests import get as request_get, post as request_post
+from requests import get as request_get, post as request_post, RequestException, Response
+from requests.exceptions import JSONDecodeError
 from time import sleep
 from .constants import JobState, NETWORK_PROTOCOLS
 
 
+class OcrdNetworkClientError(Exception):
+    """Used when requests to the processing-server fail
+
+    Requests are executed through http. If 400 or 500 status codes are returned this exception is
+    raised
+    """
+
+    def __init__(self, status_code: int, message: str) -> None:
+        self.status_code = status_code
+        self.message = message
+
+
 def _poll_endpoint_status(ps_server_host: str, job_id: str, job_type: str, tries: int, wait: int,
                           print_state: bool = False) -> JobState:
     if job_type not in ["workflow", "processor"]:
@@ -24,6 +37,18 @@ def _poll_endpoint_status(ps_server_host: str, job_id: str, job_type: str, tries
     return job_state
 
 
+def _raise_if_error(response: Response) -> None:
+    """Check the requests-response and raise an exception if its status code indicates an error"""
+    try:
+        response.raise_for_status()
+    except RequestException as e:
+        try:
+            message = response.json()["detail"]
+            raise OcrdNetworkClientError(response.status_code, message) from e
+        except JSONDecodeError:
+            raise OcrdNetworkClientError(response.status_code, response.text) from e
+
+
 def poll_job_status_till_timeout_fail_or_success(
     ps_server_host: str, job_id: str, tries: int, wait: int, print_state: bool = False) -> JobState:
     return _poll_endpoint_status(ps_server_host, job_id, "processor", tries, wait, print_state)
@@ -37,14 +62,14 @@ def poll_wf_status_till_timeout_fail_or_success(
 def get_ps_deployed_processors(ps_server_host: str):
     request_url = f"{ps_server_host}/processor"
     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}"
+    _raise_if_error(response)
     return response.json()
 
 
 def get_ps_deployed_processor_ocrd_tool(ps_server_host: str, processor_name: str):
     request_url = f"{ps_server_host}/processor/info/{processor_name}"
     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}"
+    _raise_if_error(response)
     return response.json()
 
 
@@ -57,7 +82,7 @@ def get_ps_processing_job_log(ps_server_host: str, processing_job_id: str):
 def get_ps_processing_job_status(ps_server_host: str, processing_job_id: str) -> JobState:
     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}"
+    _raise_if_error(response)
     job_state = response.json()["state"]
     assert job_state
     return getattr(JobState, job_state.lower())
@@ -66,7 +91,7 @@ def get_ps_processing_job_status(ps_server_host: str, processing_job_id: str) ->
 def get_ps_workflow_job_status(ps_server_host: str, workflow_job_id: str) -> JobState:
     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}"
+    _raise_if_error(response)
     job_state = response.json()["state"]
     assert job_state
     return getattr(JobState, job_state.lower())
@@ -79,7 +104,7 @@ def post_ps_processing_request(ps_server_host: str, processor: str, job_input: d
         headers={"accept": "application/json; charset=utf-8"},
         json=job_input
     )
-    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    _raise_if_error(response)
     processing_job_id = response.json()["job_id"]
     assert processing_job_id
     return processing_job_id
@@ -102,7 +127,7 @@ def post_ps_workflow_request(
     json_resp_raw = response.text
     # print(f'post_ps_workflow_request >> {response.status_code}')
     # print(f'post_ps_workflow_request >> {json_resp_raw}')
-    assert response.status_code == 200, f"Processing server: {request_url}, {response.status_code}"
+    _raise_if_error(response)
     wf_job_id = json.loads(json_resp_raw)["job_id"]
     assert wf_job_id
     return wf_job_id