From 592a3eb18644b8ef2cc67f15fe3681ef4ae4f4fb Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Thu, 25 Jun 2026 13:32:12 +0200 Subject: [PATCH 1/6] ogc: integrated extraction of assets and updating hrefs --- .../implementations/ogc_api_process.py | 132 ++++++++++++++---- 1 file changed, 107 insertions(+), 25 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index 546f74f..b30647b 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -27,8 +27,10 @@ "https://schemas.stacspec.org/v1.0.0/collection-spec/json-schema/collection.json" ) -GEOJSON_FEATURECOLLECTION_SCHEMA = "https://schemas.opengis.net/ogcapi/" \ +GEOJSON_FEATURECOLLECTION_SCHEMA = ( + "https://schemas.opengis.net/ogcapi/" "features/part1/1.0/openapi/schemas/featureCollectionGeoJSON.yaml" +) @register_platform(ProcessTypeEnum.OGC_API_PROCESS) @@ -292,6 +294,93 @@ def _map_ogcapi_status(self, ogcapi_status: StatusCode) -> ProcessingStatusEnum: logger.warning(f"Mapping of unknown OGC API status: {ogcapi_status}") return ProcessingStatusEnum.UNKNOWN + def _extract_download_link_from_asset(self, asset: dict) -> str | None: + """ + Extracts the download link from an asset dictionary. Checks if the `href` field is present and contains an HTTPS URL. + If this is not the case, look for an alternative link in `alternate` field. If no valid link is found, return None. + + Args: + asset (dict): The asset dictionary. + + Returns: + str | None: The download link if available, otherwise None. + """ + refs = [ + asset.get("href"), + asset.get("alternate", {}).get("https", {}).get("href"), + ] + for href in refs: + if href and href.startswith("https://"): + return href + return None + + def _generate_signed_url(self, href: str, user_token: str) -> str: + """ + Generate a signed URL for the given href using the provided user token. + This is a placeholder implementation and should be replaced with actual logic to generate signed URLs. + + Args: + href (str): The original href. + user_token (str): The user token to be used for signing. + """ + # TODO - Add implementation + logger.debug(f"Generating signed URL for href: {href} with user token.") + return href + + def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: + """ + Update the hrefs of the assets to be HTTPS URLs. If the current href is an S3 URL, the code will look into + `alternate` links to find an HTTPS URL. If no HTTPS URL is found, the original href will be kept. + """ + updated_assets = {} + for asset_name, asset in assets.items(): + updated_asset = asset.copy() + href = self._extract_download_link_from_asset(asset) + if not href: + logger.warning( + f"No valid HTTPS download link found for asset '{asset_name}'. Keeping original href. " + "Skipping asset..." + ) + else: + href = self._generate_signed_url(href, user_token) + updated_asset["href"] = href + updated_assets[asset_name] = updated_asset + + return updated_assets + + def _extract_assets_from_feature_collection( + self, feature_collection: dict, *, result_name: str, user_token: str + ) -> dict: + assets: dict = {} + for feature in feature_collection.get("features", []): + feature_assets = feature.get("assets") + if isinstance(feature_assets, dict): + assets.update(feature_assets) + continue + + # Some providers expose assets through an item link instead of inlining them in the feature. + for link in feature.get("links", []): + if "collection" == link.get("rel") and link.get("href"): + collection_link: str = link.get("href") + logger.debug( + f"GeoJSON FeatureCollection results: '{result_name}' " + f"points to a valid collection URL: {collection_link}" + ) + + response: HTTPXResponse = http_get( + collection_link, + follow_redirects=True, + headers={"Authorization": f"Bearer {user_token}"}, + ) + response.raise_for_status() + collection = Collection.model_validate(response.json()) + logger.debug( + f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}" + ) + assets.update(collection.to_dict().get("assets", {})) + break + return assets + async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails ) -> ProcessingStatusEnum: @@ -314,6 +403,7 @@ async def get_job_status( async def get_job_results( self, user_token: str, job_id: str, details: ServiceDetails ) -> Collection: + assets: dict = {} logger.debug(f"Fetching job result for opfenEO job with ID {job_id}") logger.debug("Exchanging user token for OGC API Process execution...") @@ -354,15 +444,15 @@ async def get_job_results( schema_reference = qualified_value.var_schema.actual_instance logger.debug( f"Processing result\n* Name: '{result_name}'\n" - "* media type: {qualified_value.media_type}\n" - "* Python type: {type(qualified_value.value)}\n" - "* schema {qualified_value.var_schema}..." + f"* media type: {qualified_value.media_type}\n" + f"* Python type: {type(qualified_value.value)}\n" + f"* schema {qualified_value.var_schema}..." ) if not isinstance(schema_reference, str): logger.warning( f"Processing result name: '{result_name}' can not be processed, " - "schema of type {type(schema_reference)} not recognized" + f"schema of type {type(schema_reference)} not recognized" ) continue @@ -375,29 +465,20 @@ async def get_job_results( logger.success( f"GeoJSON FeatureCollection found in results: '{result_name}'" ) - feature_collection = qualified_value.value.oneof_schema_2_validator or {} - for feature in feature_collection.get("features", []): - for link in feature.get("links", []): - if "collection" == link.get("rel") and link.get("href"): - collection_link: str = link.get("href") - logger.success( - f"GeoJSON FeatureCollection results: '{result_name}' " - "points to a valid collection URL: {collection_link}" - ) - - response: HTTPXResponse = http_get( - collection_link, - follow_redirects=True, - headers={ - "Authorization": f"Bearer {exchanged_token}" - }, - ) - response.raise_for_status() - return Collection.model_validate(response.json()) + feature_collection = ( + qualified_value.value.oneof_schema_2_validator or {} + ) + assets.update( + self._extract_assets_from_feature_collection( + feature_collection, + result_name=result_name, + user_token=user_token, + ) + ) else: logger.warning( f"Processing result: '{result_name}' can not be processed, " - "schema {schema_reference} not yet managed" + f"schema {schema_reference} not yet managed" ) # result not found, send back an empty collection @@ -417,6 +498,7 @@ async def get_job_results( spatial=SpatialExtent(bbox=[(-180.0, -90.0, 180.0, 90.0)]), temporal=TimeInterval(interval=[[None, None]]), ), + assets=self._update_assets_hrefs(assets, exchanged_token), ) async def get_service_parameters( From 0acec67d6b32fff3f193116cf505f58a906f6ed1 Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Thu, 25 Jun 2026 14:21:30 +0200 Subject: [PATCH 2/6] chore: fixed tests --- .../implementations/ogc_api_process.py | 44 ++++++++++++------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index b30647b..a0b84da 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -296,8 +296,10 @@ def _map_ogcapi_status(self, ogcapi_status: StatusCode) -> ProcessingStatusEnum: def _extract_download_link_from_asset(self, asset: dict) -> str | None: """ - Extracts the download link from an asset dictionary. Checks if the `href` field is present and contains an HTTPS URL. - If this is not the case, look for an alternative link in `alternate` field. If no valid link is found, return None. + Extracts the download link from an asset dictionary. Checks if the + `href` field is present and contains an HTTPS URL. If this is not + the case, look for an alternative link in `alternate` + field. If no valid link is found, return None. Args: asset (dict): The asset dictionary. @@ -317,7 +319,8 @@ def _extract_download_link_from_asset(self, asset: dict) -> str | None: def _generate_signed_url(self, href: str, user_token: str) -> str: """ Generate a signed URL for the given href using the provided user token. - This is a placeholder implementation and should be replaced with actual logic to generate signed URLs. + This is a placeholder implementation and should be replaced with + actual logic to generate signed URLs. Args: href (str): The original href. @@ -326,11 +329,13 @@ def _generate_signed_url(self, href: str, user_token: str) -> str: # TODO - Add implementation logger.debug(f"Generating signed URL for href: {href} with user token.") return href - + def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: """ - Update the hrefs of the assets to be HTTPS URLs. If the current href is an S3 URL, the code will look into - `alternate` links to find an HTTPS URL. If no HTTPS URL is found, the original href will be kept. + Update the hrefs of the assets to be HTTPS URLs. If the current + href is an S3 URL, the code will look into `alternate` links to + find an HTTPS URL. If no HTTPS URL is found, the original href + will be kept. """ updated_assets = {} for asset_name, asset in assets.items(): @@ -338,7 +343,8 @@ def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: href = self._extract_download_link_from_asset(asset) if not href: logger.warning( - f"No valid HTTPS download link found for asset '{asset_name}'. Keeping original href. " + "No valid HTTPS download link found for asset " + f"'{asset_name}'. Keeping original href. " "Skipping asset..." ) else: @@ -350,7 +356,7 @@ def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: def _extract_assets_from_feature_collection( self, feature_collection: dict, *, result_name: str, user_token: str - ) -> dict: + ) -> tuple[dict, Collection | None]: assets: dict = {} for feature in feature_collection.get("features", []): feature_assets = feature.get("assets") @@ -358,7 +364,8 @@ def _extract_assets_from_feature_collection( assets.update(feature_assets) continue - # Some providers expose assets through an item link instead of inlining them in the feature. + # Some providers expose assets through an item link + # instead of inlining them in the feature. for link in feature.get("links", []): if "collection" == link.get("rel") and link.get("href"): collection_link: str = link.get("href") @@ -374,12 +381,14 @@ def _extract_assets_from_feature_collection( ) response.raise_for_status() collection = Collection.model_validate(response.json()) + collection_assets = collection.assets or {} logger.debug( - f"Extracted collection '{collection.id}' with assets: {list(collection.assets.keys())}" + f"Extracted collection '{collection.id}' " + f"with assets: {list(collection_assets.keys())}" ) assets.update(collection.to_dict().get("assets", {})) - break - return assets + return assets, collection + return assets, None async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails @@ -442,9 +451,10 @@ async def get_job_results( and qualified_value.var_schema.actual_instance ): schema_reference = qualified_value.var_schema.actual_instance + media_type = getattr(qualified_value, "media_type", None) logger.debug( f"Processing result\n* Name: '{result_name}'\n" - f"* media type: {qualified_value.media_type}\n" + f"* media type: {media_type}\n" f"* Python type: {type(qualified_value.value)}\n" f"* schema {qualified_value.var_schema}..." ) @@ -468,13 +478,17 @@ async def get_job_results( feature_collection = ( qualified_value.value.oneof_schema_2_validator or {} ) - assets.update( + download_token = exchanged_token or user_token + extracted_assets, linked_collection = ( self._extract_assets_from_feature_collection( feature_collection, result_name=result_name, - user_token=user_token, + user_token=download_token, ) ) + if linked_collection: + return linked_collection + assets.update(extracted_assets) else: logger.warning( f"Processing result: '{result_name}' can not be processed, " From 4a65645d1fe0a542669c8d416477eab6a6f10920 Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Mon, 13 Jul 2026 13:15:45 +0200 Subject: [PATCH 3/6] feat: added generation of signed url --- app/platforms/implementations/ogc_api_process.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index a0b84da..e36d60f 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -1,5 +1,7 @@ import re from typing import List + +import requests from app.auth import exchange_token, get_current_user_claims from fastapi import Response from loguru import logger @@ -179,6 +181,7 @@ async def execute_job( exchanged_token = await exchange_token( user_token=user_token, url=details.endpoint ) + logger.debug(f"Exchanged token: {exchanged_token}") # Output format omitted from request api_client = await self._create_api_client_instance( @@ -327,8 +330,11 @@ def _generate_signed_url(self, href: str, user_token: str) -> str: user_token (str): The user token to be used for signing. """ # TODO - Add implementation - logger.debug(f"Generating signed URL for href: {href} with user token.") - return href + logger.debug(f"Generating signed URL for href: {href} with user token.") + response = requests.get(href, headers={"Authorization": f"Bearer {user_token}"}, allow_redirects=False) + signed_url = response.headers["location"] + logger.debug(f"Signed URL: {signed_url}") + return signed_url def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: """ From 7b237c9508a7eb549b41c4c849eaee3e825c2d0d Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Tue, 14 Jul 2026 14:30:56 +0200 Subject: [PATCH 4/6] chore: cleanup of debug statement --- app/platforms/implementations/ogc_api_process.py | 1 - 1 file changed, 1 deletion(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index e36d60f..647bda3 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -181,7 +181,6 @@ async def execute_job( exchanged_token = await exchange_token( user_token=user_token, url=details.endpoint ) - logger.debug(f"Exchanged token: {exchanged_token}") # Output format omitted from request api_client = await self._create_api_client_instance( From e1f7ea1abf43decf0cc24e255d47e4d56540666c Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Tue, 14 Jul 2026 16:59:08 +0200 Subject: [PATCH 5/6] feat: integrated result visualization of OGC API --- .../implementations/ogc_api_process.py | 143 +++++++++++++++--- 1 file changed, 120 insertions(+), 23 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index 647bda3..b907df3 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -1,3 +1,4 @@ +import json import re from typing import List @@ -329,8 +330,12 @@ def _generate_signed_url(self, href: str, user_token: str) -> str: user_token (str): The user token to be used for signing. """ # TODO - Add implementation - logger.debug(f"Generating signed URL for href: {href} with user token.") - response = requests.get(href, headers={"Authorization": f"Bearer {user_token}"}, allow_redirects=False) + logger.debug(f"Generating signed URL for href: {href} with user token.") + response = requests.get( + href, + headers={"Authorization": f"Bearer {user_token}"}, + allow_redirects=False, + ) signed_url = response.headers["location"] logger.debug(f"Signed URL: {signed_url}") return signed_url @@ -359,11 +364,98 @@ def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict: return updated_assets + def _build_collection_from_features( + self, + features: list, + assets: dict, + result_name: str, + user_token: str, + details: ServiceDetails, + internal_job_id: str, + ) -> Collection: + """ + Build a STAC Collection from a list of GeoJSON features and their + aggregated assets. The spatial extent is derived from the feature + bounding boxes and the temporal extent from the feature datetime + properties. + + Args: + features: GeoJSON feature list. + assets: Aggregated asset dict collected from those features. + result_name: Identifier used as the collection ID. + user_token: Token used to sign asset hrefs. + details: Service details containing namespace and application information. + internal_job_id: Internal job identifier. + + Returns: + A STAC Collection. + """ + # Spatial extent — union of per-feature bboxes + min_x, min_y = float("inf"), float("inf") + max_x, max_y = float("-inf"), float("-inf") + found_bbox = False + for feature in features: + bbox = feature.get("bbox") + if isinstance(bbox, (list, tuple)) and len(bbox) >= 4: + min_x = min(min_x, float(bbox[0])) + min_y = min(min_y, float(bbox[1])) + max_x = max(max_x, float(bbox[2])) + max_y = max(max_y, float(bbox[3])) + found_bbox = True + spatial_bbox = ( + [min_x, min_y, max_x, max_y] if found_bbox else [-180.0, -90.0, 180.0, 90.0] + ) + + # Temporal extent — min/max of all datetime-like properties + datetimes: list[str] = [] + for feature in features: + props = feature.get("properties") or {} + for dt_key in ("datetime", "start_datetime", "end_datetime"): + dt_val = props.get(dt_key) + if isinstance(dt_val, str): + datetimes.append(dt_val) + temporal_interval: list[list] = ( + [[min(datetimes), max(datetimes)]] if datetimes else [[None, None]] + ) + + updated_assets = self._update_assets_hrefs(assets, user_token) + + logger.debug( + f"Building STAC Collection '{result_name}' from {len(features)} feature(s) " + f"with {len(updated_assets)} asset(s)." + ) + + return Collection( + id=f"{details.namespace}-{internal_job_id}", + stac_version=STAC_VERSION, + title=f"Results for {details.application}", + description=( + f"OGC API process result items for job '{internal_job_id}' " + f"of application '{details.application}'." + ), + type="Collection", + license="proprietary", + links=Links([]), + extent=Extent( + spatial=SpatialExtent(bbox=[spatial_bbox]), + temporal=TimeInterval(interval=temporal_interval), + ), + assets=updated_assets, + ) + def _extract_assets_from_feature_collection( - self, feature_collection: dict, *, result_name: str, user_token: str - ) -> tuple[dict, Collection | None]: + self, + feature_collection: dict, + *, + result_name: str, + user_token: str, + details: ServiceDetails, + internal_job_id: str, + ) -> Collection: assets: dict = {} - for feature in feature_collection.get("features", []): + features = feature_collection.get("features", []) + logger.debug(f"Feature collection: {json.dumps(feature_collection, indent=2)}") + for feature in features: feature_assets = feature.get("assets") if isinstance(feature_assets, dict): assets.update(feature_assets) @@ -385,15 +477,25 @@ def _extract_assets_from_feature_collection( headers={"Authorization": f"Bearer {user_token}"}, ) response.raise_for_status() - collection = Collection.model_validate(response.json()) - collection_assets = collection.assets or {} + collection_data = response.json() + collection_data["assets"] = self._update_assets_hrefs( + collection_data.get("assets", {}), user_token + ) + collection = Collection.model_validate(collection_data) logger.debug( f"Extracted collection '{collection.id}' " - f"with assets: {list(collection_assets.keys())}" + f"with assets: {list((collection.assets or {}).keys())}" ) - assets.update(collection.to_dict().get("assets", {})) - return assets, collection - return assets, None + return collection + + return self._build_collection_from_features( + features, + assets, + result_name, + user_token, + details, + internal_job_id + ) async def get_job_status( self, user_token: str, job_id: str, details: ServiceDetails @@ -417,7 +519,6 @@ async def get_job_status( async def get_job_results( self, user_token: str, job_id: str, details: ServiceDetails ) -> Collection: - assets: dict = {} logger.debug(f"Fetching job result for opfenEO job with ID {job_id}") logger.debug("Exchanging user token for OGC API Process execution...") @@ -483,17 +584,13 @@ async def get_job_results( feature_collection = ( qualified_value.value.oneof_schema_2_validator or {} ) - download_token = exchanged_token or user_token - extracted_assets, linked_collection = ( - self._extract_assets_from_feature_collection( - feature_collection, - result_name=result_name, - user_token=download_token, - ) + return self._extract_assets_from_feature_collection( + feature_collection, + result_name=result_name, + user_token=exchanged_token or user_token, + details=details, + internal_job_id=internal_job_id, ) - if linked_collection: - return linked_collection - assets.update(extracted_assets) else: logger.warning( f"Processing result: '{result_name}' can not be processed, " @@ -517,7 +614,7 @@ async def get_job_results( spatial=SpatialExtent(bbox=[(-180.0, -90.0, 180.0, 90.0)]), temporal=TimeInterval(interval=[[None, None]]), ), - assets=self._update_assets_hrefs(assets, exchanged_token), + assets={}, ) async def get_service_parameters( From 9b40893714aa0d9fd6528bfed6a77283292e08f6 Mon Sep 17 00:00:00 2001 From: bramjanssen Date: Tue, 14 Jul 2026 17:20:36 +0200 Subject: [PATCH 6/6] chore: fixed linting --- app/platforms/implementations/ogc_api_process.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index b907df3..d751b7f 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/app/platforms/implementations/ogc_api_process.py @@ -402,8 +402,8 @@ def _build_collection_from_features( max_x = max(max_x, float(bbox[2])) max_y = max(max_y, float(bbox[3])) found_bbox = True - spatial_bbox = ( - [min_x, min_y, max_x, max_y] if found_bbox else [-180.0, -90.0, 180.0, 90.0] + spatial_bbox: tuple[float, float, float, float] = ( + (min_x, min_y, max_x, max_y) if found_bbox else (-180.0, -90.0, 180.0, 90.0) ) # Temporal extent — min/max of all datetime-like properties