From 01070bc058b19ba78f6a1a9b56df0e7d459a9bd2 Mon Sep 17 00:00:00 2001 From: stripodi Date: Tue, 20 Jan 2026 12:23:02 +0100 Subject: [PATCH 1/3] feat: ogc-api-processes-client initial plug --- .../implementations/ogc_api_process.py | 24 ++++++++++++++++++- requirements.txt | 3 ++- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index 620a3b5..d55daa7 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -6,7 +6,10 @@ from app.schemas.parameters import Parameter from app.schemas.unit_job import ServiceDetails from stac_pydantic import Collection - +from ogc_api_client import Configuration +from ogc_api_client.api.execute_api import ExecuteApi +from ogc_api_client.api_client_wrapper import ApiClientWrapper +from ogc_api_client.rest import ApiException @register_platform(ProcessTypeEnum.OGC_API_PROCESS) class OGCAPIProcessPlatform(BaseProcessingPlatform): @@ -15,6 +18,15 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform): This class handles the execution of processing jobs on the OGC API Process platform. """ + def _create_api_client_instance( + self, + details: ServiceDetails + ) -> ApiClientWrapper: + configuration: Configuration = Configuration( + host = details.endpoint + ) + return ApiClientWrapper(configuration) + async def execute_job( self, user_token: str, @@ -23,6 +35,8 @@ async def execute_job( parameters: dict, format: OutputFormatEnum, ) -> str: + api_client = self._create_api_client_instance(details) + raise NotImplementedError("OGC API Process job execution not implemented yet.") async def execute_synchronous_job( @@ -33,11 +47,15 @@ async def execute_synchronous_job( parameters: dict, format: OutputFormatEnum, ) -> Response: + # This is currently not supported + raise NotImplementedError("OGC API Process job execution not implemented yet.") async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails ) -> ProcessingStatusEnum: + api_client = self._create_api_client_instance(details) + raise NotImplementedError( "OGC API Process job status retrieval not implemented yet." ) @@ -45,6 +63,8 @@ async def get_job_status( async def get_job_results( self, user_token: str, job_id: str, details: ServiceDetails ) -> Collection: + api_client = self._create_api_client_instance(details) + raise NotImplementedError( "OGC API Process job result retrieval not implemented yet." ) @@ -52,6 +72,8 @@ async def get_job_results( async def get_service_parameters( self, user_token: str, details: ServiceDetails ) -> List[Parameter]: + api_client = self._create_api_client_instance(details) + raise NotImplementedError( "OGC API Process service parameter retrieval not implemented yet." ) diff --git a/requirements.txt b/requirements.txt index 115e5b6..e8d92ef 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,4 +29,5 @@ SQLAlchemy stac-pydantic types-requests types-shapely -uvicorn[standard] \ No newline at end of file +uvicorn[standard] +ogc-api-client @ git+https://github.com/EOEPCA/ogc-api-client.git@develop#subdirectory=src From aaaa9259d0cac964b31d6ad01b42b52fd0d29504 Mon Sep 17 00:00:00 2001 From: floeschau Date: Wed, 18 Feb 2026 17:21:51 +0100 Subject: [PATCH 2/3] Changes for OGC API platform --- .../implementations/ogc_api_process.py | 134 +++++++++++++++--- app/schemas/parameters.py | 3 + app/schemas/unit_job.py | 8 ++ 3 files changed, 129 insertions(+), 16 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index d55daa7..3f7a4ef 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -1,5 +1,10 @@ +import json + +import re from typing import List from fastapi import Response +from loguru import logger + from app.platforms.base import BaseProcessingPlatform from app.platforms.dispatcher import register_platform from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum, ProcessingStatusEnum @@ -10,22 +15,41 @@ from ogc_api_client.api.execute_api import ExecuteApi from ogc_api_client.api_client_wrapper import ApiClientWrapper from ogc_api_client.rest import ApiException +from ogc_api_client.models.status_info import StatusInfo, StatusCode @register_platform(ProcessTypeEnum.OGC_API_PROCESS) class OGCAPIProcessPlatform(BaseProcessingPlatform): + + application_path_regex = re.compile(r"(?P.+)/processes/(?P[^/]+)$") + """ OGC API Process processing platform implementation. This class handles the execution of processing jobs on the OGC API Process platform. """ + def _split_job_id(self, job_id): + parts = job_id.split(":", 1) + if len(parts) != 2: + return (None, job_id) + return tuple(parts) + + def _create_api_client_instance( self, - details: ServiceDetails + endpoint: str, + namespace: str, + user_token: str = None, ) -> ApiClientWrapper: configuration: Configuration = Configuration( - host = details.endpoint + host = f"{endpoint}/{namespace}" if namespace else endpoint ) - return ApiClientWrapper(configuration) + + additional_args = {} + if user_token: + additional_args["header_name"] = "Authorization" + additional_args["header_value"] = f"Bearer {user_token}" + + return ApiClientWrapper(configuration, **additional_args) async def execute_job( self, @@ -35,9 +59,32 @@ async def execute_job( parameters: dict, format: OutputFormatEnum, ) -> str: - api_client = self._create_api_client_instance(details) + logger.info(f"Executing OGC API job with title={title}") + # Output format omitted from request + + api_client = self._create_api_client_instance(details.endpoint, details.namespace, user_token) + + headers = { + "accept": "*/*", + #"Prefer": "respond-async;return=representation", + "Content-Type": "application/json" + } + if user_token: + headers["Authorization"] = f"Bearer {user_token}" + + data = { + "inputs": {key: value for key, value in parameters.items()} + } + + content = api_client.execute_simple(process_id=details.application, execute=data, _headers=headers) + + job_id = content.job_id + + # Return the namespace along with the job ID if needed + if details.namespace: + return f"{details.namespace}:{job_id}" + return job_id - raise NotImplementedError("OGC API Process job execution not implemented yet.") async def execute_synchronous_job( self, @@ -51,29 +98,84 @@ async def execute_synchronous_job( raise NotImplementedError("OGC API Process job execution not implemented yet.") + + def _map_ogcapi_status(self, ogcapi_status: str) -> ProcessingStatusEnum: + """ + Map the status returned by OGC API to a status known within the API. + + :param status: Status text returned by OGC API. + :return: ProcessingStatusEnum corresponding to the input. + """ + + logger.debug(f"Mapping OGC API status {ogcapi_status} to ProcessingStatusEnum") + + mapping = { + StatusCode.ACCEPTED: ProcessingStatusEnum.CREATED, + StatusCode.RUNNING: ProcessingStatusEnum.RUNNING, + StatusCode.DISMISSED: ProcessingStatusEnum.CANCELED, + StatusCode.SUCCESSFUL: ProcessingStatusEnum.FINISHED, + StatusCode.FAILED: ProcessingStatusEnum.FAILED, + } + + try: + return mapping[ogcapi_status] + except (AttributeError, KeyError): + logger.warning("Mapping of unknown OGC API status: %r", ogcapi_status) + return ProcessingStatusEnum.UNKNOWN + + + async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails ) -> ProcessingStatusEnum: - api_client = self._create_api_client_instance(details) + logger.debug(f"Fetching job status for OGC API job with ID {job_id}") + + # Job ID is composed of namespace and internal job id + namespace, internal_job_id = self._split_job_id(job_id) + api_client = self._create_api_client_instance(details.endpoint, namespace, user_token) + + status_info = api_client.get_status(job_id=internal_job_id) + return self._map_ogcapi_status(status_info.status) - raise NotImplementedError( - "OGC API Process job status retrieval not implemented yet." - ) async def get_job_results( self, user_token: str, job_id: str, details: ServiceDetails ) -> Collection: - api_client = self._create_api_client_instance(details) + logger.debug(f"Fetching job result for opfenEO job with ID {job_id}") + + # Job ID is composed of namespace and internal job id + namespace, internal_job_id = self._split_job_id(job_id) + api_client = self._create_api_client_instance(details.endpoint, namespace, user_token) + + result = api_client.get_result(job_id=internal_job_id) + return Collection(result[0]) - raise NotImplementedError( - "OGC API Process job result retrieval not implemented yet." - ) async def get_service_parameters( self, user_token: str, details: ServiceDetails ) -> List[Parameter]: - api_client = self._create_api_client_instance(details) - raise NotImplementedError( - "OGC API Process service parameter retrieval not implemented yet." + parameters = [] + logger.debug( + f"Fetching service parameters for OGC API process with ID {details.application}" ) + + api_client = self._create_api_client_instance(details.endpoint, details.namespace, user_token) + process_description = api_client.get_process_description(details.application) + + for input_id, input_details in process_description.inputs.items(): + input_type = input_id, input_details.model_dump().get("var_schema", {}).get("actual_instance", {}).get("type", "string") + if isinstance(input_type, tuple): + input_type = next((t for t in input_type if t in ["date-interval", "bounding-box", "boolean"]), "string") + + parameters.append( + Parameter( + name=input_id, + description=input_details.description, + default=None, + optional=(input_details.min_occurs == 0), + type="string", + ) + ) + + return parameters diff --git a/app/schemas/parameters.py b/app/schemas/parameters.py index 04e6f46..0e4c286 100644 --- a/app/schemas/parameters.py +++ b/app/schemas/parameters.py @@ -7,9 +7,12 @@ class ParamTypeEnum(str, Enum): + DATETIME = "datetime" DATE_INTERVAL = "date-interval" BOUNDING_BOX = "bounding-box" BOOLEAN = "boolean" + INTEGER = "integer" + DOUBLE = "double" STRING = "string" diff --git a/app/schemas/unit_job.py b/app/schemas/unit_job.py index 2bc997f..5194576 100644 --- a/app/schemas/unit_job.py +++ b/app/schemas/unit_job.py @@ -1,6 +1,7 @@ from datetime import datetime from pydantic import BaseModel, Field +from typing import Optional from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum, ProcessTypeEnum @@ -13,6 +14,13 @@ class ServiceDetails(BaseModel): "platform API", examples=["https://openeofed.dataspace.copernicus.eu"], ) + namespace: Optional[str] = Field( + default=None, + description="Namespace under the endpoint where the service is hosted. For openEO, this field" + "is not set. For OGC API Processes, this field should include the namespace ID representing" + "under which the namespace-related API is deployed", + examples=["https://openeofed.dataspace.copernicus.eu"] + ) application: str = Field( ..., description="Path to the application that needs to be executed. For openEO this is " From 010fb9ec8b9842fba9a3c3ad4770a1565aa46741 Mon Sep 17 00:00:00 2001 From: floeschau Date: Wed, 11 Mar 2026 13:11:48 +0100 Subject: [PATCH 3/3] Format (ogc_api_process.py) --- .../implementations/ogc_api_process.py | 66 ++++++++++++------- 1 file changed, 41 insertions(+), 25 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index 3f7a4ef..5addb70 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -17,10 +17,12 @@ from ogc_api_client.rest import ApiException from ogc_api_client.models.status_info import StatusInfo, StatusCode + @register_platform(ProcessTypeEnum.OGC_API_PROCESS) class OGCAPIProcessPlatform(BaseProcessingPlatform): - - application_path_regex = re.compile(r"(?P.+)/processes/(?P[^/]+)$") + application_path_regex = re.compile( + r"(?P.+)/processes/(?P[^/]+)$" + ) """ OGC API Process processing platform implementation. @@ -33,7 +35,6 @@ def _split_job_id(self, job_id): return (None, job_id) return tuple(parts) - def _create_api_client_instance( self, endpoint: str, @@ -41,7 +42,7 @@ def _create_api_client_instance( user_token: str = None, ) -> ApiClientWrapper: configuration: Configuration = Configuration( - host = f"{endpoint}/{namespace}" if namespace else endpoint + host=f"{endpoint}/{namespace}" if namespace else endpoint ) additional_args = {} @@ -62,30 +63,31 @@ async def execute_job( logger.info(f"Executing OGC API job with title={title}") # Output format omitted from request - api_client = self._create_api_client_instance(details.endpoint, details.namespace, user_token) + api_client = self._create_api_client_instance( + details.endpoint, details.namespace, user_token + ) headers = { "accept": "*/*", - #"Prefer": "respond-async;return=representation", - "Content-Type": "application/json" + # "Prefer": "respond-async;return=representation", + "Content-Type": "application/json", } if user_token: headers["Authorization"] = f"Bearer {user_token}" - data = { - "inputs": {key: value for key, value in parameters.items()} - } - - content = api_client.execute_simple(process_id=details.application, execute=data, _headers=headers) + data = {"inputs": {key: value for key, value in parameters.items()}} + + content = api_client.execute_simple( + process_id=details.application, execute=data, _headers=headers + ) job_id = content.job_id - + # Return the namespace along with the job ID if needed if details.namespace: return f"{details.namespace}:{job_id}" return job_id - async def execute_synchronous_job( self, user_token: str, @@ -98,7 +100,6 @@ async def execute_synchronous_job( raise NotImplementedError("OGC API Process job execution not implemented yet.") - def _map_ogcapi_status(self, ogcapi_status: str) -> ProcessingStatusEnum: """ Map the status returned by OGC API to a status known within the API. @@ -123,21 +124,20 @@ def _map_ogcapi_status(self, ogcapi_status: str) -> ProcessingStatusEnum: logger.warning("Mapping of unknown OGC API status: %r", ogcapi_status) return ProcessingStatusEnum.UNKNOWN - - async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails ) -> ProcessingStatusEnum: logger.debug(f"Fetching job status for OGC API job with ID {job_id}") - + # Job ID is composed of namespace and internal job id namespace, internal_job_id = self._split_job_id(job_id) - api_client = self._create_api_client_instance(details.endpoint, namespace, user_token) + api_client = self._create_api_client_instance( + details.endpoint, namespace, user_token + ) status_info = api_client.get_status(job_id=internal_job_id) return self._map_ogcapi_status(status_info.status) - async def get_job_results( self, user_token: str, job_id: str, details: ServiceDetails ) -> Collection: @@ -145,12 +145,13 @@ async def get_job_results( # Job ID is composed of namespace and internal job id namespace, internal_job_id = self._split_job_id(job_id) - api_client = self._create_api_client_instance(details.endpoint, namespace, user_token) + api_client = self._create_api_client_instance( + details.endpoint, namespace, user_token + ) result = api_client.get_result(job_id=internal_job_id) return Collection(result[0]) - async def get_service_parameters( self, user_token: str, details: ServiceDetails ) -> List[Parameter]: @@ -160,13 +161,28 @@ async def get_service_parameters( f"Fetching service parameters for OGC API process with ID {details.application}" ) - api_client = self._create_api_client_instance(details.endpoint, details.namespace, user_token) + api_client = self._create_api_client_instance( + details.endpoint, details.namespace, user_token + ) process_description = api_client.get_process_description(details.application) for input_id, input_details in process_description.inputs.items(): - input_type = input_id, input_details.model_dump().get("var_schema", {}).get("actual_instance", {}).get("type", "string") + input_type = ( + input_id, + input_details.model_dump() + .get("var_schema", {}) + .get("actual_instance", {}) + .get("type", "string"), + ) if isinstance(input_type, tuple): - input_type = next((t for t in input_type if t in ["date-interval", "bounding-box", "boolean"]), "string") + input_type = next( + ( + t + for t in input_type + if t in ["date-interval", "bounding-box", "boolean"] + ), + "string", + ) parameters.append( Parameter(