Review

- breaking up the code in smaller modules
- improving error handling
- adding job queue inspection methods

Author: pacesm

eoxs_wps_async/backend.py

@@ -4,7 +4,7 @@
 #
 # Authors: Martin Paces <martin.paces@eox.at>
 #-------------------------------------------------------------------------------
-# Copyright (C) 2016 EOX IT Services GmbH
+# Copyright (C) 2016-2023 EOX IT Services GmbH
 #
 # Permission is hereby granted, free of charge, to any person obtaining a copy
 # of this software and associated documentation files (the "Software"), to deal
@@ -26,85 +26,53 @@
 #-------------------------------------------------------------------------------
 # pylint: disable=unused-argument, too-many-arguments, no-self-use
 
-from uuid import uuid4
 from logging import getLogger
-from eoxserver.services.ows.wps.exceptions import ServerBusy, NoApplicableCode
 from eoxs_wps_async.util import cached_property
 from eoxs_wps_async.config import get_wps_config
 from eoxs_wps_async.client import Client, ClientError
-from eoxs_wps_async.handler import (
-    check_job_id, get_job_logger, get_response_url, get_response,
-    is_valid_job_id,
-)
+from eoxs_wps_async.protocol import ClientProtocol
+from eoxs_wps_async.handler import get_response_url, get_response
 
 LOGGER_NAME = "eoxserver.services.ows.wps"
 
 
-class WPSAsyncBackend():
+class WPSAsyncBackend:
     """ WPS asynchronous back-end. """
     supported_versions = ("1.0.0",)
 
     def execute(self, process, raw_inputs, resp_form, extra_parts=None,
                 job_id=None, version="1.0.0", **kwargs):
         """ Asynchronous process execution. """
-        job_id = check_job_id(job_id or str(uuid4()))
-        logger = get_job_logger(job_id, LOGGER_NAME)
-
-        response, *payload = self._request(
-            "EXECUTE",
-            job_id,
-            process.identifier,
-            raw_inputs,
-            resp_form,
-            extra_parts
-        )
-
-        if response == "OK":
-            return job_id
-        if response == "BUSY":
-            raise ServerBusy("The server is busy!")
-        if response == "OWSEXC":
-            raise payload[0]
-        if response == "ERROR":
-            raise NoApplicableCode(payload[0], "eoxs_wps_async.daemon")
-
-        return self._handle_unknown_response(response, logger)
+        self._send_request(self._protocol.execute_request(
+            job_id, process.identifier, raw_inputs, resp_form, extra_parts,
+        ))
+        return job_id
 
     def purge(self, job_id, process_id=None, **kwargs):
         """ Purge the job from the system by removing all the resources
         occupied by the job.
         If the optional process_id is provided then the discard() callback
         of the process is executed.
         """
-        job_id = check_job_id(job_id)
-        logger = get_job_logger(job_id, LOGGER_NAME)
-
-        response, *payload = self._request("PURGE", job_id, process_id)
-
-        if response == "OK":
-            return
-        if response == "ERROR":
-            raise ClientError(payload[0])
-
-        self._handle_unknown_response(response, logger)
+        self._send_request(self._protocol.purge_request(job_id, process_id))
 
     def list(self, job_ids=None, **kwargs):
         """ List current jobs. The list can be restricted by the given job_ids.
         """
-        logger = getLogger(LOGGER_NAME)
-
-        if job_ids is not None:
-            # convert to a list and reject invalid job ids
-            job_ids = [id_ for id_ in job_ids if is_valid_job_id(id_)]
+        job_ids, *_ = self._send_request(self._protocol.list_request(job_ids))
+        return job_ids
 
-        response, *payload = self._request("LIST", job_ids)
-
-        if response == "OK":
-            return payload[0]
-        if response == "ERROR":
-            raise ClientError(payload[0])
+    def list_queue(self, job_ids=None, **kwargs):
+        """ List queued jobs. The list can be restricted by the given job_ids.
+        """
+        jobs, *_ = self._send_request(self._protocol.list_queue_request(job_ids))
+        return jobs
 
-        return self._handle_unknown_response(response, logger)
+    def get_queue_size(self, **kwargs):
+        """ Get number of queued jobs.
+        """
+        size, *_ = self._send_request(self._protocol.get_queue_size_request())
+        return size
 
     def get_response_url(self, job_id):
         """ Return response URL for the given job identifier. """
@@ -116,11 +84,28 @@ def get_response(self, job_id):
         """
         return get_response(job_id, self._conf)
 
+    def _send_request(self, request, logger=None):
+        """ Send request and handle response. """
+
+        def _request(request):
+            with self._client as client:
+                client.send(request)
+                return client.recv()
+
+        return self._protocol.handle_response(
+            _request(request), logger=logger
+        )
+
     @cached_property
     def _conf(self):
         """ Get configuration. """
         return get_wps_config()
 
+    @cached_property
+    def _protocol(self):
+        """ Get client protocol instance. """
+        return ClientProtocol(getLogger(LOGGER_NAME))
+
     @property
     def _client(self):
         """ Get connection to the execution daemon. """
@@ -137,16 +122,6 @@ def _client(self):
             self._conf.socket_connection_timeout,
         )
 
-    def _request(self, *request):
-        with self._client as client:
-            client.send(request)
-            return client.recv()
-
-    def _handle_unknown_response(self, response, logger):
-        message = f"Unknown response! RESP={response!r}"
-        logger.error(message)
-        raise ValueError(message)
-
 
 # alias for backward compatibility
 WPSAsyncBackendBase = WPSAsyncBackend