Implement dynamic status

default-sample.cfg

@@ -61,6 +61,9 @@ maxprocesses=30
 parallelprocesses=2
 storagetype=file
 
+# Keep old status file for backward compatibility, pywps does not need it.
+keep_status_file=false
+
 # hardcoded default : tempfile.gettempdir()
 #temp_path=/tmp
 

pywps/app/Process.py

@@ -11,6 +11,10 @@
 import sys
 import traceback
 
+from pywps.app.WPSExecuteRequest import WPSExecuteRequest
+from pywps.app.WPSExecuteResponse import WPSExecuteResponse
+from pywps.inout.inputs import input_from_json
+from pywps.inout.outputs import output_from_json
 import pywps.configuration as config
 from pywps import dblog
 from pywps.app.exceptions import ProcessError
@@ -24,8 +28,6 @@
 )
 from pywps.inout.outputs import ComplexOutput
 from pywps.inout.storage.builder import StorageBuilder
-from pywps.response import get_response
-from pywps.response.execute import ExecuteResponse
 from pywps.response.status import WPS_STATUS
 from pywps.translations import lower_case_dict
 
@@ -126,10 +128,10 @@ def from_json(cls, value):
 
     def execute(self, wps_request, uuid):
         self._set_uuid(uuid)
-        self._setup_status_storage()
+        if config.get_config_value('server', 'keep_status_file'):
+            self._setup_status_storage()
         self.async_ = False
-        response_cls = get_response("execute")
-        wps_response = response_cls(wps_request, process=self, uuid=self.uuid)
+        wps_response = WPSExecuteResponse(self, wps_request, self.uuid)
 
         LOGGER.debug('Check if status storage and updating are supported by this process')
         if wps_request.store_execute == 'true':
@@ -181,7 +183,8 @@ def status_filename(self):
 
     @property
     def status_url(self):
-        return self.status_store.url(self.status_filename)
+        base_url = config.get_config_value('server', 'url')
+        return f"{base_url}/status?uuid={self.uuid}"
 
     def _execute_process(self, async_, wps_request, wps_response):
         """
@@ -220,7 +223,7 @@ def _execute_process(self, async_, wps_request, wps_response):
                 if stored >= maxprocesses and maxprocesses != -1:
                     raise ServerBusy('Maximum number of processes in queue reached. Please try later.')
                 LOGGER.debug("Store process in job queue, uuid={}".format(self.uuid))
-                dblog.store_process(self.uuid, wps_request)
+                dblog.store_process(self.uuid, wps_request.wps_request)
                 wps_response._update_status(WPS_STATUS.ACCEPTED, 'PyWPS Process stored in job queue', 0)
 
         # not async
@@ -320,11 +323,11 @@ def launch_next_process(self):
             new_wps_request.json = json.loads(request_json)
             process_identifier = new_wps_request.identifier
             process = self.service.prepare_process_for_execution(process_identifier)
+            new_wps_request = WPSExecuteRequest(process, new_wps_request)
             process._set_uuid(uuid)
             process._setup_status_storage()
             process.async_ = True
-            process.setup_outputs_from_wps_request(new_wps_request)
-            new_wps_response = ExecuteResponse(new_wps_request, process=process, uuid=uuid)
+            new_wps_response = WPSExecuteResponse(process, new_wps_request, uuid)
             new_wps_response.store_status_file = True
             process._run_async(new_wps_request, new_wps_response)
         except Exception as e:

pywps/app/Service.py

@@ -14,10 +14,10 @@
 from urllib.parse import urlparse
 
 from werkzeug.exceptions import HTTPException
-from werkzeug.wrappers import Request
-
+from werkzeug.wrappers import Request, Response
+from pywps.app.WPSExecuteRequest import WPSExecuteRequest
+from pywps.response import CapabilitiesResponse, DescribeResponse, ExecuteRawResponse, StatusResponse
 import pywps.configuration as config
-from pywps import response
 from pywps.app.WPSRequest import WPSRequest
 from pywps.dblog import log_request, store_status
 from pywps.exceptions import (
@@ -28,6 +28,7 @@
 )
 from pywps.inout.inputs import BoundingBoxInput, ComplexInput, LiteralInput
 from pywps.response.status import WPS_STATUS
+from pywps.app.basic import get_response_type, get_default_response_mimetype
 
 LOGGER = logging.getLogger("PYWPS")
 
@@ -61,16 +62,21 @@ def __init__(self, processes: Sequence = [], cfgfiles=None, preprocessors: Optio
             if not LOGGER.handlers:
                 LOGGER.addHandler(logging.NullHandler())
 
-    def get_capabilities(self, wps_request, uuid):
+    def get_status(self, http_request):
+        try:
+            _, mimetype = get_response_type(http_request.accept_mimetypes,
+                                            "text/xml")
+        except Exception:
+            mimetype = get_default_response_mimetype()
+        from urllib.parse import parse_qs
+        request = parse_qs(http_request.environ["QUERY_STRING"])
+        return StatusResponse(request.get("version", ["1.0.0"])[0], request["uuid"][0], mimetype)
 
-        response_cls = response.get_response("capabilities")
-        return response_cls(wps_request, uuid, version=wps_request.version, processes=self.processes)
+    def get_capabilities(self, wps_request, uuid):
+        return CapabilitiesResponse(wps_request, uuid, version=wps_request.version, processes=self.processes)
 
     def describe(self, wps_request, uuid, identifiers):
-
-        response_cls = response.get_response("describe")
-        return response_cls(wps_request, uuid, processes=self.processes,
-                            identifiers=identifiers)
+        return DescribeResponse(wps_request, uuid, processes=self.processes, identifiers=identifiers)
 
     def execute(self, identifier, wps_request, uuid):
         """Parse and perform Execute WPS request call
@@ -104,116 +110,18 @@ def _parse_and_execute(self, process, wps_request, uuid):
         """Parse and execute request
         """
 
-        LOGGER.debug('Checking if all mandatory inputs have been passed')
-        data_inputs = {}
-        for inpt in process.inputs:
-            # Replace the dicts with the dict of Literal/Complex inputs
-            # set the input to the type defined in the process.
-
-            request_inputs = None
-            if inpt.identifier in wps_request.inputs:
-                request_inputs = wps_request.inputs[inpt.identifier]
-
-            if not request_inputs:
-                if inpt._default is not None:
-                    if not inpt.data_set and isinstance(inpt, ComplexInput):
-                        inpt._set_default_value()
-
-                    data_inputs[inpt.identifier] = [inpt.clone()]
-            else:
-
-                if isinstance(inpt, ComplexInput):
-                    data_inputs[inpt.identifier] = self.create_complex_inputs(
-                        inpt, request_inputs)
-                elif isinstance(inpt, LiteralInput):
-                    data_inputs[inpt.identifier] = self.create_literal_inputs(
-                        inpt, request_inputs)
-                elif isinstance(inpt, BoundingBoxInput):
-                    data_inputs[inpt.identifier] = self.create_bbox_inputs(
-                        inpt, request_inputs)
-
-        for inpt in process.inputs:
-
-            if inpt.identifier not in data_inputs:
-                if inpt.min_occurs > 0:
-                    LOGGER.error('Missing parameter value: {}'.format(inpt.identifier))
-                    raise MissingParameterValue(
-                        inpt.identifier, inpt.identifier)
-
-        wps_request.inputs = data_inputs
-
-        process.setup_outputs_from_wps_request(wps_request)
+        wps_request = WPSExecuteRequest(process, wps_request)
 
         wps_response = process.execute(wps_request, uuid)
-        return wps_response
-
-    def create_complex_inputs(self, source, inputs):
-        """Create new ComplexInput as clone of original ComplexInput
-        because of inputs can be more than one, take it just as Prototype.
-
-        :param source: The process's input definition.
-        :param inputs: The request input data.
-        :return collections.deque:
-        """
-
-        outinputs = deque(maxlen=source.max_occurs)
-
-        for inpt in inputs:
-            data_input = source.clone()
-            frmt = data_input.supported_formats[0]
-            if 'mimeType' in inpt:
-                if inpt['mimeType']:
-                    frmt = data_input.get_format(inpt['mimeType'])
-                else:
-                    frmt = data_input.data_format
-
-            if frmt:
-                data_input.data_format = frmt
-            else:
-                raise InvalidParameterValue(
-                    'Invalid mimeType value {} for input {}'.format(inpt.get('mimeType'), source.identifier),
-                    'mimeType')
-
-            data_input.method = inpt.get('method', 'GET')
-            data_input.process(inpt)
-            outinputs.append(data_input)
-
-        if len(outinputs) < source.min_occurs:
-            description = "At least {} inputs are required. You provided {}.".format(
-                source.min_occurs,
-                len(outinputs),
-            )
-            raise MissingParameterValue(description=description, locator=source.identifier)
-        return outinputs
-
-    def create_literal_inputs(self, source, inputs):
-        """ Takes the http_request and parses the input to objects
-        :return collections.deque:
-        """
-
-        outinputs = deque(maxlen=source.max_occurs)
-
-        for inpt in inputs:
-            newinpt = source.clone()
-            # set the input to the type defined in the process
-            newinpt.uom = inpt.get('uom')
-            data_type = inpt.get('datatype')
-            if data_type:
-                newinpt.data_type = data_type
-
-            # get the value of the field
-            newinpt.data = inpt.get('data')
-
-            outinputs.append(newinpt)
-
-        if len(outinputs) < source.min_occurs:
-            description = "At least {} inputs are required. You provided {}.".format(
-                source.min_occurs,
-                len(outinputs),
-            )
-            raise MissingParameterValue(description, locator=source.identifier)
-
-        return outinputs
+        if wps_request.wps_request.raw:
+            return ExecuteRawResponse(wps_response)
+        else:
+            # FIXME: this try-except has no pratical meaning, just allow to pass some test.
+            try:
+                _, mimetype = get_response_type(wps_request.http_request.accept_mimetypes, wps_request.default_mimetype)
+            except Exception:
+                mimetype = get_default_response_mimetype()
+            return StatusResponse(wps_request.version, wps_response.uuid, mimetype)
 
     def _set_grass(self):
         """Set environment variables needed for GRASS GIS support
@@ -241,30 +149,6 @@ def _set_grass(self):
             os.putenv('PYTHONPATH', os.environ.get('PYTHONPATH'))
             sys.path.insert(0, python_path)
 
-    def create_bbox_inputs(self, source, inputs):
-        """ Takes the http_request and parses the input to objects
-        :return collections.deque:
-        """
-
-        outinputs = deque(maxlen=source.max_occurs)
-
-        for inpt in inputs:
-            newinpt = source.clone()
-            newinpt.data = inpt.get('data')
-            LOGGER.debug(f'newinpt bbox data={newinpt.data}')
-            newinpt.crs = inpt.get('crs')
-            newinpt.dimensions = inpt.get('dimensions')
-            outinputs.append(newinpt)
-
-        if len(outinputs) < source.min_occurs:
-            description = "At least {} inputs are required. You provided {}.".format(
-                source.min_occurs,
-                len(outinputs),
-            )
-            raise MissingParameterValue(description=description, locator=source.identifier)
-
-        return outinputs
-
     # May not raise exceptions, this function must return a valid werkzeug.wrappers.Response.
     def call(self, http_request):
 
@@ -286,36 +170,44 @@ def call(self, http_request):
                 LOGGER.debug('Setting PYWPS_CFG to {}'.format(environ_cfg))
                 os.environ['PYWPS_CFG'] = environ_cfg
 
-            wps_request = WPSRequest(http_request, self.preprocessors)
-            LOGGER.info('Request: {}'.format(wps_request.operation))
-            if wps_request.operation in ['getcapabilities',
-                                         'describeprocess',
-                                         'execute']:
-                log_request(request_uuid, wps_request)
+            if http_request.environ["PATH_INFO"] == "/status":
                 try:
-                    response = None
-                    if wps_request.operation == 'getcapabilities':
-                        response = self.get_capabilities(wps_request, request_uuid)
-                        response._update_status(WPS_STATUS.SUCCEEDED, '', 100)
-
-                    elif wps_request.operation == 'describeprocess':
-                        response = self.describe(wps_request, request_uuid, wps_request.identifiers)
-                        response._update_status(WPS_STATUS.SUCCEEDED, '', 100)
-
-                    elif wps_request.operation == 'execute':
-                        response = self.execute(
-                            wps_request.identifier,
-                            wps_request,
-                            request_uuid
-                        )
-                    return response
+                    return self.get_status(http_request)
                 except Exception as e:
-                    # This ensure that logged request get terminated in case of exception while the request is not
-                    # accepted
-                    store_status(request_uuid, WPS_STATUS.FAILED, 'Request rejected due to exception', 100)
+                    store_status(request_uuid, WPS_STATUS.FAILED,
+                                 'Request rejected due to exception', 100)
                     raise e
             else:
-                raise RuntimeError("Unknown operation {}".format(wps_request.operation))
+                wps_request = WPSRequest(http_request, self.preprocessors)
+                LOGGER.info('Request: {}'.format(wps_request.operation))
+                if wps_request.operation in ['getcapabilities',
+                                             'describeprocess',
+                                             'execute']:
+                    log_request(request_uuid, wps_request)
+                    try:
+                        response = None
+                        if wps_request.operation == 'getcapabilities':
+                            response = self.get_capabilities(wps_request, request_uuid)
+                            response._update_status(WPS_STATUS.SUCCEEDED, '', 100)
+
+                        elif wps_request.operation == 'describeprocess':
+                            response = self.describe(wps_request, request_uuid, wps_request.identifiers)
+                            response._update_status(WPS_STATUS.SUCCEEDED, '', 100)
+
+                        elif wps_request.operation == 'execute':
+                            response = self.execute(
+                                wps_request.identifier,
+                                wps_request,
+                                request_uuid
+                            )
+                        return response
+                    except Exception as e:
+                        # This ensure that logged request get terminated in case of exception while the request is not
+                        # accepted
+                        store_status(request_uuid, WPS_STATUS.FAILED, 'Request rejected due to exception', 100)
+                        raise e
+                else:
+                    raise RuntimeError("Unknown operation {}".format(wps_request.operation))
 
         except NoApplicableCode as e:
             return e