Skip to content

Commit b501918

Browse files
JanssenBrmJanssenBrm
authored
Merge pull request #68 from ESA-APEx/feature/ogc-api-input-fix
Improve OGC API Process payload metadata, parameter mapping, and test coverage
2 parents 1e08394 + bf75b22 commit b501918

5 files changed

Lines changed: 1080 additions & 52 deletions

File tree

‎app/auth.py‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,8 +49,11 @@ def _decode_token(token: str):
4949

5050

5151
def get_current_user_id(token: str = Depends(oauth2_scheme)):
52-
user: dict = _decode_token(token)
53-
return user["sub"]
52+
return get_current_user_claims(token)["sub"]
53+
54+
55+
def get_current_user_claims(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
56+
return _decode_token(token)
5457

5558

5659
async def websocket_authenticate(websocket: WebSocket) -> str | None:

‎app/platforms/implementations/ogc_api_process.py‎

Lines changed: 145 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import re
22
from typing import List
3-
from app.auth import exchange_token
3+
from app.auth import exchange_token, get_current_user_claims
44
from fastapi import Response
55
from loguru import logger
66

@@ -34,11 +34,14 @@
3434
@register_platform(ProcessTypeEnum.OGC_API_PROCESS)
3535
class OGCAPIProcessPlatform(BaseProcessingPlatform):
3636
input_type_map = {
37+
"date-time": ParamTypeEnum.DATETIME,
3738
"date-interval": ParamTypeEnum.DATE_INTERVAL,
3839
"bounding-box": ParamTypeEnum.BOUNDING_BOX,
3940
"boolean": ParamTypeEnum.BOOLEAN,
4041
"integer": ParamTypeEnum.INTEGER,
4142
"double": ParamTypeEnum.DOUBLE,
43+
"number": ParamTypeEnum.DOUBLE,
44+
"string": ParamTypeEnum.STRING,
4245
}
4346

4447
status_mapping = {
@@ -53,6 +56,12 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform):
5356
r"(?P<namespace>.+)/processes/(?P<process_id>[^/]+)$"
5457
)
5558

59+
geojson_schema_references = {
60+
GEOJSON_FEATURECOLLECTION_SCHEMA,
61+
"https://geojson.org/schema/FeatureCollection.json",
62+
"https://geojson.org/schema/Feature.json",
63+
}
64+
5665
"""
5766
OGC API Process processing platform implementation.
5867
This class handles the execution of processing jobs on the OGC API Process platform.
@@ -64,6 +73,76 @@ def _split_job_id(self, job_id) -> tuple[str, ...]:
6473
return ("", job_id)
6574
return tuple(parts)
6675

76+
def _get_type_from_schema(
77+
self, schema: dict | str | None, input_id: str = ""
78+
) -> ParamTypeEnum:
79+
if isinstance(schema, str):
80+
if schema in self.__class__.geojson_schema_references:
81+
return ParamTypeEnum.POLYGON
82+
return self.__class__.input_type_map.get(schema, ParamTypeEnum.STRING)
83+
84+
if not isinstance(schema, dict):
85+
return ParamTypeEnum.STRING
86+
87+
schema_type = str(schema.get("type"))
88+
schema_format = schema.get("format")
89+
schema_subtype = schema.get("subtype")
90+
91+
if schema_type == "array" and schema_subtype == "date-interval":
92+
return ParamTypeEnum.DATE_INTERVAL
93+
if schema_type == "array" and schema.get("items", {}).get("type") == "string":
94+
return ParamTypeEnum.ARRAY_STRING
95+
if schema_subtype == "geojson":
96+
return ParamTypeEnum.POLYGON
97+
if schema_subtype == "bounding-box":
98+
return ParamTypeEnum.BOUNDING_BOX
99+
if schema_format == "geojson":
100+
return ParamTypeEnum.POLYGON
101+
if schema_format == "date-time":
102+
return ParamTypeEnum.DATETIME
103+
if schema_type == "object":
104+
required = schema.get("required") or []
105+
if "type" in required and "coordinates" in required:
106+
type_properties = schema.get("properties", {}).get("type", {})
107+
type_instance = type_properties
108+
while "actual_instance" in type_instance:
109+
type_instance = type_instance["actual_instance"]
110+
if "Polygon" in type_instance.get("enum", []):
111+
return ParamTypeEnum.POLYGON
112+
if "Point" in type_instance.get("enum", []):
113+
return ParamTypeEnum.POINT
114+
elif "bbox" in required:
115+
return ParamTypeEnum.BOUNDING_BOX
116+
117+
if isinstance(schema.get("$ref"), str):
118+
return self._get_type_from_schema(schema.get("$ref"), input_id)
119+
120+
for variant_key in ("oneOf", "anyOf", "allOf"):
121+
variants = schema.get(variant_key) or []
122+
if not isinstance(variants, list):
123+
continue
124+
for variant in variants:
125+
detected_type = self._get_type_from_schema(variant, input_id)
126+
if detected_type != ParamTypeEnum.STRING:
127+
return detected_type
128+
129+
properties = schema.get("properties") or {}
130+
if (
131+
schema.get("title") == "GeoJSON"
132+
or "geometry" in properties
133+
or "features" in properties
134+
or input_id.lower() in {"aoi", "geometry", "geom", "geojson"}
135+
):
136+
return ParamTypeEnum.POLYGON
137+
138+
return self.__class__.input_type_map.get(schema_type, ParamTypeEnum.STRING)
139+
140+
def _get_options_from_schema(self, schema: dict | str | None) -> list:
141+
if not isinstance(schema, dict):
142+
return []
143+
options = schema.get("enum")
144+
return options if isinstance(options, list) else []
145+
67146
async def _create_api_client_instance(
68147
self,
69148
endpoint: str,
@@ -91,6 +170,8 @@ async def execute_job(
91170
) -> str:
92171
logger.info(f"Executing OGC API job with title={title}")
93172

173+
parameters = await self._transform_parameters(user_token, details, parameters)
174+
94175
# Exchanging token
95176
logger.debug("Exchanging user token for OGC API Process execution...")
96177
exchanged_token = await exchange_token(
@@ -112,7 +193,22 @@ async def execute_job(
112193
if exchanged_token:
113194
headers["Authorization"] = f"Bearer {exchanged_token}"
114195

115-
data = {"inputs": {key: value for key, value in parameters.items()}}
196+
user_claims = get_current_user_claims(user_token)
197+
properties = {
198+
"title": title,
199+
"application": details.application,
200+
}
201+
if user_claims.get("sub"):
202+
properties["user_id"] = user_claims["sub"]
203+
if user_claims.get("preferred_username"):
204+
properties["username"] = user_claims["preferred_username"]
205+
if user_claims.get("email"):
206+
properties["email"] = user_claims["email"]
207+
208+
data = {
209+
"inputs": parameters,
210+
"properties": properties,
211+
}
116212

117213
content = api_client.execute_simple(
118214
process_id=details.application, execute=data, _headers=headers
@@ -125,6 +221,49 @@ async def execute_job(
125221
return f"{details.namespace}:{job_id}"
126222
return job_id
127223

224+
def _transform_bbox_parameter(self, param_name: str, value) -> list[float]:
225+
if isinstance(value, (list, tuple)) and len(value) == 4:
226+
return [float(coord) for coord in value]
227+
228+
if isinstance(value, dict):
229+
if ["east", "north", "south", "west"] == sorted(value.keys()):
230+
return [
231+
float(value["west"]),
232+
float(value["south"]),
233+
float(value["east"]),
234+
float(value["north"]),
235+
]
236+
237+
raise ValueError(
238+
f"Unsupported bounding box value for parameter {param_name}: {value}"
239+
)
240+
241+
async def _transform_parameters(
242+
self, user_token: str, details: ServiceDetails, parameters: dict
243+
) -> dict:
244+
service_params = await self.get_service_parameters(user_token, details)
245+
246+
modifiers = {
247+
ParamTypeEnum.BOUNDING_BOX: self._transform_bbox_parameter,
248+
}
249+
250+
transformed_parameters = parameters.copy()
251+
for param in service_params:
252+
if param.name not in parameters:
253+
continue
254+
255+
modifier = modifiers.get(param.type)
256+
257+
if modifier:
258+
transformed_parameters[param.name] = modifier(
259+
param.name, parameters[param.name]
260+
)
261+
262+
logger.debug(
263+
f"Transformed parameters for OGC API Process: {transformed_parameters}"
264+
)
265+
return transformed_parameters
266+
128267
async def execute_synchronous_job(
129268
self,
130269
user_token: str,
@@ -303,53 +442,10 @@ async def get_service_parameters(
303442

304443
if process_description.inputs:
305444
for input_id, input_details in process_description.inputs.items():
306-
input_type = (
307-
input_id,
308-
input_details.model_dump()
309-
.get("var_schema", {})
310-
.get("actual_instance", {})
311-
.get("type", ""),
312-
)
313-
if isinstance(input_type, tuple):
314-
input_type_str = next(
315-
(
316-
t
317-
for t in input_type
318-
if t
319-
in [
320-
"date-interval",
321-
"bounding-box",
322-
"boolean",
323-
"integer",
324-
"double",
325-
]
326-
),
327-
None,
328-
)
329-
else:
330-
input_type_str = None
331-
332-
if input_type_str:
333-
input_type_str = self.__class__.input_type_map.get(input_type_str)
334-
335-
if not input_type_str:
336-
input_type_str = ParamTypeEnum.STRING
337-
input_types = (
338-
input_details.model_dump()
339-
.get("var_schema", {})
340-
.get("actual_instance", {})
341-
.get("required")
342-
or []
343-
)
344-
if "bbox" in input_types:
345-
input_type_str = ParamTypeEnum.BOUNDING_BOX
346-
347-
input_options = (
445+
schema = (
348446
input_details.model_dump()
349447
.get("var_schema", {})
350-
.get("actual_instance", {})
351-
.get("enum")
352-
or []
448+
.get("actual_instance")
353449
)
354450
parameters.append(
355451
Parameter(
@@ -359,8 +455,8 @@ async def get_service_parameters(
359455
else f"Parameter: {input_id}",
360456
default=None,
361457
optional=(input_details.min_occurs == 0),
362-
type=input_type_str,
363-
options=input_options,
458+
type=self._get_type_from_schema(schema, input_id),
459+
options=self._get_options_from_schema(schema),
364460
)
365461
)
366462

‎app/schemas/parameters.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ class ParamTypeEnum(str, Enum):
1111
DATE_INTERVAL = "date-interval"
1212
BOUNDING_BOX = "bounding-box"
1313
POLYGON = "polygon"
14+
POINT = "point"
1415
BOOLEAN = "boolean"
1516
INTEGER = "integer"
1617
DOUBLE = "double"

0 commit comments

Comments
 (0)