Process manager generate job ids (geopython#1209)

* Modified process manager to generate job id

* Continued to refactor in order to let process manager generate job id

* fix syntax error

* reorder imports

Author: ricardogsilva

pygeoapi/api.py

@@ -48,7 +48,6 @@
 import re
 from typing import Any, Tuple, Union, Optional
 import urllib.parse
-import uuid
 
 from dateutil.parser import parse as dateparse
 from pygeofilter.parsers.ecql import parse as parse_ecql_text
@@ -3545,11 +3544,6 @@ def execute_process(self, request: Union[APIRequest, Any],
         data_dict = data.get('inputs', {})
         LOGGER.debug(data_dict)
 
-        job_id = data.get("job_id", str(uuid.uuid1()))
-        url = f"{self.base_url}/jobs/{job_id}"
-
-        headers['Location'] = url
-
         is_async = data.get('mode', 'auto') == 'async'
         if is_async:
             LOGGER.debug('Asynchronous request mode detected')
@@ -3560,8 +3554,9 @@ def execute_process(self, request: Union[APIRequest, Any],
 
         try:
             LOGGER.debug('Executing process')
-            mime_type, outputs, status = self.manager.execute_process(
-                process, job_id, data_dict, is_async)
+            job_id, mime_type, outputs, status = self.manager.execute_process(
+                process, data_dict, is_async)
+            headers['Location'] = f"{self.base_url}/jobs/{job_id}"
         except ProcessorExecuteError as err:
             LOGGER.error(err)
             msg = 'Processing error'

pygeoapi/process/manager/base.py

@@ -33,9 +33,10 @@
 from multiprocessing import dummy
 from pathlib import Path
 from typing import Any, Tuple
+import uuid
 
-from pygeoapi.util import DATETIME_FORMAT, JobStatus
 from pygeoapi.process.base import BaseProcessor
+from pygeoapi.util import DATETIME_FORMAT, JobStatus
 
 LOGGER = logging.getLogger(__name__)
 
@@ -129,7 +130,7 @@ def delete_job(self, job_id: str) -> bool:
         raise NotImplementedError()
 
     def _execute_handler_async(self, p: BaseProcessor, job_id: str,
-                               data_dict: dict) -> Tuple[None, JobStatus]:
+                               data_dict: dict) -> Tuple[str, None, JobStatus]:
         """
         This private execution handler executes a process in a background
         thread using `multiprocessing.dummy`
@@ -259,24 +260,30 @@ def _execute_handler_sync(self, p: BaseProcessor, job_id: str,
 
         return jfmt, outputs, current_status
 
-    def execute_process(self, p, job_id, data_dict, is_async=False):
+    def execute_process(
+            self,
+            p,
+            data_dict,
+            is_async=False
+    ) -> Tuple[str, str, Any, JobStatus]:
         """
         Default process execution handler
 
         :param p: `pygeoapi.process` object
-        :param job_id: job identifier
         :param data_dict: `dict` of data parameters
         :param is_async: `bool` specifying sync or async processing.
 
-        :returns: tuple of MIME type, response payload and status
+        :returns: tuple of job_id, MIME type, response payload and status
         """
 
+        job_id = str(uuid.uuid1())
         if not is_async:
             LOGGER.debug('Synchronous execution')
-            return self._execute_handler_sync(p, job_id, data_dict)
+            result = self._execute_handler_sync(p, job_id, data_dict)
         else:
             LOGGER.debug('Asynchronous execution')
-            return self._execute_handler_async(p, job_id, data_dict)
+            result = self._execute_handler_async(p, job_id, data_dict)
+        return (job_id, *result)
 
     def __repr__(self):
         return f'<BaseManager> {self.name}'

pygeoapi/process/manager/dummy.py

@@ -29,6 +29,7 @@
 
 import logging
 from typing import Any, Tuple
+import uuid
 
 from pygeoapi.process.base import BaseProcessor
 from pygeoapi.process.manager.base import BaseManager
@@ -63,17 +64,20 @@ def get_jobs(self, status: JobStatus = None) -> list:
 
         return []
 
-    def execute_process(self, p: BaseProcessor, job_id: str, data_dict: dict,
-                        is_async: bool = False) -> Tuple[str, Any, int]:
+    def execute_process(
+            self,
+            p: BaseProcessor,
+            data_dict: dict,
+            is_async: bool = False
+    ) -> Tuple[str, str, Any, JobStatus]:
         """
         Default process execution handler
 
         :param p: `pygeoapi.process` object
-        :param job_id: job identifier
         :param data_dict: `dict` of data parameters
         :param is_async: `bool` specifying sync or async processing.
 
-        :returns: tuple of MIME type, response payload and status
+        :returns: tuple of job_id, MIME type, response payload and status
         """
 
         jfmt = 'application/json'
@@ -93,7 +97,8 @@ def execute_process(self, p: BaseProcessor, job_id: str, data_dict: dict,
             current_status = JobStatus.failed
             LOGGER.error(err)
 
-        return jfmt, outputs, current_status
+        job_id = str(uuid.uuid1())
+        return job_id, jfmt, outputs, current_status
 
     def __repr__(self):
         return f'<DummyManager> {self.name}'