diff --git a/CHANGELOG.md b/CHANGELOG.md index 67dfeeb1..9a7ce769 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,12 @@ and start a new "In Progress" section above it. -## In progress: 0.140.0 +## In progress: 0.141.0 + +- Extract openeo-geopyspark-driver oriented batch job result metadata handling from generic view layer and allow openeo-aggregator to inject an alternative implementation in context of "STAC 1.1" batch job result metadata style ([Open-EO/openeo-aggregator/204](https://github.com/Open-EO/openeo-aggregator/issues/204)) + + +## 0.140.0 - Include `job_options` as top-level properties in `GET /jobs/{job_id}` response ([#470](https://github.com/Open-EO/openeo-python-driver/issues/470)) - Bump STAC version from `0.9.0` to `1.0.0` in capabilities endpoint, collection metadata, job results and ML model metadata ([#363](https://github.com/Open-EO/openeo-python-driver/issues/363)) diff --git a/openeo_driver/_version.py b/openeo_driver/_version.py index 677a947a..e64b243e 100644 --- a/openeo_driver/_version.py +++ b/openeo_driver/_version.py @@ -1 +1 @@ -__version__ = "0.140.0a8" +__version__ = "0.141.0a1" diff --git a/openeo_driver/backend.py b/openeo_driver/backend.py index 54676a57..26a7e445 100644 --- a/openeo_driver/backend.py +++ b/openeo_driver/backend.py @@ -490,6 +490,7 @@ def to_api_dict(self, full=True, api_version: ComparableVersion = None) -> dict: @dataclasses.dataclass(frozen=True) class BatchJobResultMetadata: # Basic dataclass based wrapper for batch job result metadata (allows cleaner code navigation and discovery) + # TODO: a flat top-level asset list is a deprecated concept, migrate to a proper 'collection -> items -> asset' hierachy assets: Dict[str, dict] = dataclasses.field(default_factory=dict) items: Dict[str, dict] = dataclasses.field(default_factory=dict) links: List[dict] = dataclasses.field(default_factory=list) @@ -629,9 +630,11 @@ def start_job(self, job_id: str, user: User): def get_result_metadata(self, job_id: str, user_id: str) -> BatchJobResultMetadata: """ - Get job result metadata + High-level API to list batch job results in an opinionated, structured way, + (e.g. targeting the `GET /jobs/{job_id}/results` endpoint). - https://openeo.org/documentation/1.0/developers/api/reference.html#tag/Batch-Jobs/operation/list-results + Also see `list_job_results` for a more low-level API + that returns a raw, less-opinionated metadata document directly. """ # Default implementation, based on existing components return BatchJobResultMetadata( @@ -662,6 +665,37 @@ def _get_providers(self, job_id: str, user_id: str): } ] + def list_job_results( + self, *, job_id: str, user_id: str, partial: bool = False, api_version: ComparableVersion + ) -> dict: + """ + Low-level API to list batch job results, + targeting the `GET /jobs/{job_id}/results` endpoint. + + Returns a raw batch job results metadata document. + + Also see `get_result_metadata` for a higher-level API + that returns a more opinionated, structured result. + """ + # TODO: this is the legacy openeo-geopyspark-driver oriented implementation + # ideally this can be migrated to GpsBatchJobs + from openeo_driver.backend_._geopyspark import list_job_results + + return list_job_results( + batch_jobs=self, job_id=job_id, user_id=user_id, partial=partial, api_version=api_version + ) + + def get_item_metadata_doc(self, *, job_id: str, user_id: str, item_id: str, format: Optional[str] = None) -> dict: + """ + Low-level API to return a batch job result item metadata document, + e.g. targetting endpoints like `GET /jobs//results/items/` + """ + # TODO: this is the legacy openeo-geopyspark-driver oriented implementation + # ideally this can be migrated to GpsBatchJobs + from openeo_driver.backend_._geopyspark import get_item_metadata_doc + + return get_item_metadata_doc(batch_jobs=self, job_id=job_id, item_id=item_id, user_id=user_id, format=format) + def get_result_assets(self, job_id: str, user_id: str) -> Dict[str, dict]: """ Return result assets as (filename, metadata) mapping: `filename` is the part that @@ -671,7 +705,7 @@ def get_result_assets(self, job_id: str, user_id: str) -> Dict[str, dict]: related: https://openeo.org/documentation/1.0/developers/api/reference.html#tag/Batch-Jobs/operation/list-results """ - # Default implementation, based on legacy API + # TODO: deprecate/phase-out this outdated, 0.4-style API return self.get_results(job_id=job_id, user_id=user_id) def get_results(self, job_id: str, user_id: str) -> Dict[str, dict]: diff --git a/openeo_driver/backend_/__init__.py b/openeo_driver/backend_/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openeo_driver/backend_/_geopyspark.py b/openeo_driver/backend_/_geopyspark.py new file mode 100644 index 00000000..d90e8676 --- /dev/null +++ b/openeo_driver/backend_/_geopyspark.py @@ -0,0 +1,852 @@ +""" + +The initial goal of this module was to extract +the existing, but openeo-geopyspark-driver oriented +batch job result metadata handling from the generic view layer, +to allow openeo-aggregator to inject an alternative implementation +of the batch job result metadata handling, without having +to wrestle geopyspark-driver specific quirks in the view layer. + +This code is still kept in openeo-python-driver for now to ease the migration path. +Ideally, in the long term however, most of this should probably +be moved to openeo-geopyspark-driver. + +""" + + +import copy +import logging +import os +import urllib.parse +from pathlib import Path +from typing import List, Optional + +import flask +import openeo +import openeo.metadata +from openeo.util import Rfc3339, TimingLogger, deep_get, dict_no_none, rfc3339 +from openeo.utils.version import ComparableVersion + +from openeo_driver.backend import ( + BatchJobMetadata, + BatchJobResultMetadata, + BatchJobs, +) +from openeo_driver.views_.batch_jobs import list_job_results_add_basic_links +from openeo_driver.config import get_backend_config +from openeo_driver.constants import ( + ITEM_LINK_PROPERTY, + JOB_STATUS, + STAC_EXTENSION, + STAC_ITEM_MEDIA_TYPE, +) +from openeo_driver.datacube import DriverMlModel +from openeo_driver.errors import FilePathInvalidException, JobNotFinishedException, OpenEOApiException +from openeo_driver.jobregistry import PARTIAL_JOB_STATUS +from openeo_driver.users import user_id_b64_encode +from openeo_driver.util.geometry import BoundingBox +from openeo_driver.util.stac import sniff_stac_extension_prefix + + +_log = logging.getLogger(__name__) + + +def list_job_results( + *, batch_jobs: BatchJobs, job_id: str, user_id: str, partial: bool = False, api_version: ComparableVersion +) -> dict: + # TODO: this is the legacy openeo-geopyspark-driver oriented implementation + + with TimingLogger(f"backend_implementation.batch_jobs.get_job_info({job_id=}, {user_id=})", logger=_log): + job_info = batch_jobs.get_job_info(job_id, user_id) + + if job_info.status != JOB_STATUS.FINISHED: + if not partial: + raise JobNotFinishedException() + else: + return _list_job_results_partial(user_id=user_id, job_id=job_id, job_info=job_info, partial=partial) + + with TimingLogger(f"backend_implementation.batch_jobs.get_result_metadata({job_id=}, {user_id=})", logger=_log): + result_metadata = batch_jobs.get_result_metadata(job_id=job_id, user_id=user_id) + + if api_version.at_least("1.1.0"): + if result_metadata.items: + # "STAC 1.1" style result listing (STAC Collection with focus on item-level assets) + return _list_job_results_stac11( + user_id=user_id, job_id=job_id, job_info=job_info, result_metadata=result_metadata + ) + else: + # "openEO 1.1.0" style result listing (STAC Collection with focus on collection-level assets) + return _list_job_results_openeo110( + user_id=user_id, job_id=job_id, job_info=job_info, result_metadata=result_metadata + ) + else: + # "openEO 1.0.0" style result listing (STAC Item) + _log.warning(f"Using old STAC Item style job result listing for {job_id=} ({api_version=})") + return _list_job_results_openeo100( + user_id=user_id, job_id=job_id, job_info=job_info, result_metadata=result_metadata + ) + + +def _list_job_results_partial(*, user_id: str, job_id: str, job_info: BatchJobMetadata, partial: bool) -> dict: + links = list_job_results_add_basic_links( + links=[], job_id=job_id, user_id=user_id, partial=partial, add_card4l=False + ) + result = { + "openeo:status": PARTIAL_JOB_STATUS.for_job_status(job_info.status), + "type": "Collection", + "stac_version": "1.0.0", + "id": job_id, + "title": job_info.title or f"Unfinished batch job {job_id}", + "description": job_info.description or f"Results for batch job {job_id}", + "license": "proprietary", # TODO? + "extent": { + "spatial": {"bbox": [[-180, -90, 180, 90]]}, + "temporal": {"interval": [[rfc3339.now_utc(), rfc3339.now_utc()]]}, + }, + "links": links, + } + return result + + + + + + +def _job_result_item_url(*, job_id: str, item_id: str, user_id: str, is11: bool = False) -> str: + signer = get_backend_config().url_signer + + method_start = ".get_job_result_item" + if is11: + method_start = method_start + "11" + if not signer: + return flask.url_for(method_start, job_id=job_id, item_id=item_id, _external=True) + + expires = signer.get_expires() + secure_key = signer.sign_job_item(job_id=job_id, user_id=user_id, item_id=item_id, expires=expires) + user_base64 = user_id_b64_encode(user_id) + return flask.url_for( + method_start + "_signed", + job_id=job_id, + user_base64=user_base64, + secure_key=secure_key, + item_id=item_id, + expires=expires, + _external=True, + ) + + +def _list_job_results_stac11( + *, + user_id: str, + job_id: str, + job_info: BatchJobMetadata, + result_metadata: BatchJobResultMetadata, +) -> dict: + """ + Batch job result listing in "STAC1.1" style: + a STAC collection, collection-level assets are deprecated in favor of item-level assets, + asset keys should not be assumed to be filenames + """ + to_datetime = Rfc3339(propagate_none=True).datetime + + links: List[dict] = copy.deepcopy(result_metadata.links or job_info.links or []) + links = list_job_results_add_basic_links(links=links, job_id=job_id, user_id=user_id) + + def intersect_band_array(list1, list2): + band_result = [] + for item1 in list1: + if isinstance(item1, dict) and "name" in item1: + for item2 in list2: + if isinstance(item1, dict) and "name" in item1 and item1["name"] == item2["name"]: + band_result.append(intersect_dicts(item1, item2)) + return band_result + + def intersect_dicts(dict1, dict2): + result = {} + for key in dict1: + if key in dict2: + if isinstance(dict1[key], dict) and isinstance(dict2[key], dict): + # Recursively intersect nested dictionaries + nested_result = intersect_dicts(dict1[key], dict2[key]) + if nested_result: # Only add if the nested result is not empty + result[key] = nested_result + elif isinstance(dict1[key], list) and isinstance(dict2[key], list) and key == "bands": + result[key] = intersect_band_array(dict1[key], dict2[key]) + elif dict1[key] == dict2[key]: + # Retain the key-value pair if values are equal + result[key] = dict1[key] + return result + + item_assets = {} + assets = {} + for item_key, item_metadata in result_metadata.items.items(): + for asset_key, asset_metadata in item_metadata.get("assets", {}).items(): + if "output_dir" in asset_metadata: + out_dir = asset_metadata.get("output_dir") + _log.info(f"asset has output dir {out_dir} and href {asset_metadata.get('href')}") + common = os.path.commonpath([asset_metadata.get("href"), out_dir]) + href = os.path.relpath(asset_metadata.get("href"), common) + else: + href = asset_metadata.get("href") + asset_object = _asset_object( + job_id=job_id, + user_id=user_id, + filename=href, + asset_metadata=asset_metadata, + job_info=job_info, + stac11=True, + ) + assets[item_key + "_" + asset_key] = asset_object + item_asset = dict_no_none( + { + "type": asset_object.get("type"), + "roles": asset_object.get("roles"), + "bands": asset_object.get("bands"), + "proj:bbox": asset_object.get("proj:bbox"), + "proj:epsg": asset_object.get("proj:epsg"), + "proj:shape": asset_object.get("proj:shape"), + "file:size": asset_object.get("file:size"), + } + ) + if asset_key not in item_assets: + item_assets[asset_key] = item_asset + else: + item_assets[asset_key] = intersect_dicts(item_assets[asset_key], item_asset) + for item_id in result_metadata.items.keys(): + links.append( + { + "rel": "item", + "href": _job_result_item_url(job_id=job_id, item_id=item_id, user_id=user_id, is11=True), + "type": STAC_ITEM_MEDIA_TYPE, + } + ) + stac_version = "1.1.0" + + links = [_normalize_job_result_link(link=k, job_id=job_id, user_id=user_id) for k in links] + + result = dict_no_none( + { + "type": "Collection", + "stac_version": stac_version, + "stac_extensions": [ + STAC_EXTENSION.EO_V110, + STAC_EXTENSION.FILEINFO, + STAC_EXTENSION.PROCESSING, + STAC_EXTENSION.PROJECTION_V120, + ], + "id": job_id, + "title": job_info.title, + "description": job_info.description or f"Results for batch job {job_id}", + "license": "proprietary", # TODO? + "extent": { + "spatial": {"bbox": [job_info.bbox] if job_info.bbox else [[-180, -90, 180, 90]]}, + "temporal": {"interval": [[to_datetime(job_info.start_datetime), to_datetime(job_info.end_datetime)]]}, + }, + "summaries": {"instruments": job_info.instruments} if job_info.instruments else {}, + "providers": result_metadata.providers or None, + "links": links, + "assets": assets, + "item_assets": item_assets, + "openeo:status": PARTIAL_JOB_STATUS.FINISHED, + } + ) + return result + + +def _list_job_results_openeo110( + *, + user_id: str, + job_id: str, + job_info: BatchJobMetadata, + result_metadata: BatchJobResultMetadata, +) -> dict: + """ + Batch job result listing in "openEO API 1.1.0, but pre-STAC1.1" style: + a STAC collection, but with focus on collection-level assets + (with filenames as asset keys) + """ + to_datetime = Rfc3339(propagate_none=True).datetime + ml_model_metadata = None + + links: List[dict] = copy.deepcopy(result_metadata.links or job_info.links or []) + links = list_job_results_add_basic_links(links=links, job_id=job_id, user_id=user_id) + + assets = { + filename: _asset_object( + job_id=job_id, + user_id=user_id, + filename=filename, + asset_metadata=asset_metadata, + job_info=job_info, + stac11=False, + ) + for filename, asset_metadata in result_metadata.assets.items() + if asset_metadata.get("asset", True) + } + + item_assets = None + for filename, metadata in result_metadata.assets.items(): + if "data" in metadata.get("roles", []) and any( + media_type in metadata.get("type", "") + for media_type in ["geotiff", "netcdf", "text/csv", "application/parquet"] + ): + links.append( + { + "rel": "item", + "href": _job_result_item_url(job_id=job_id, item_id=filename, user_id=user_id), + "type": STAC_ITEM_MEDIA_TYPE, + } + ) + elif metadata.get("ml_model_metadata", False): + # TODO: Currently we only support one ml_model per batch job. + ml_model_metadata = metadata + links.append( + { + "rel": "item", + "href": _job_result_item_url(job_id=job_id, item_id=filename, user_id=user_id), + "type": "application/json", + } + ) + stac_version = "1.0.0" + + links = [_normalize_job_result_link(link=k, job_id=job_id, user_id=user_id) for k in links] + + result = dict_no_none( + { + "type": "Collection", + "stac_version": stac_version, + "stac_extensions": [ + STAC_EXTENSION.EO_V110, + STAC_EXTENSION.FILEINFO, + STAC_EXTENSION.PROCESSING, + STAC_EXTENSION.PROJECTION_V120, + ], + "id": job_id, + "title": job_info.title, + "description": job_info.description or f"Results for batch job {job_id}", + "license": "proprietary", # TODO? + "extent": { + "spatial": {"bbox": [job_info.bbox] if job_info.bbox else [[-180, -90, 180, 90]]}, + "temporal": {"interval": [[to_datetime(job_info.start_datetime), to_datetime(job_info.end_datetime)]]}, + }, + "summaries": {"instruments": job_info.instruments} if job_info.instruments else {}, + "providers": result_metadata.providers or None, + "links": links, + "assets": assets, + "item_assets": item_assets, + "openeo:status": PARTIAL_JOB_STATUS.FINISHED, + } + ) + + if ml_model_metadata is not None: + result["stac_extensions"].extend(ml_model_metadata.get("stac_extensions", [])) + if "summaries" not in result.keys(): + result["summaries"] = {} + if "properties" in ml_model_metadata.keys(): + ml_model_properties = ml_model_metadata["properties"] + learning_approach = ml_model_properties.get("ml-model:learning_approach", None) + prediction_type = ml_model_properties.get("ml-model:prediction_type", None) + architecture = ml_model_properties.get("ml-model:architecture", None) + result["summaries"].update( + { + "ml-model:learning_approach": [learning_approach] if learning_approach is not None else [], + "ml-model:prediction_type": [prediction_type] if prediction_type is not None else [], + "ml-model:architecture": [architecture] if architecture is not None else [], + } + ) + return result + + +def _list_job_results_openeo100( + *, + user_id: str, + job_id: str, + job_info: BatchJobMetadata, + result_metadata: BatchJobResultMetadata, +) -> dict: + """ + Batch job result listing in deprecated "openEO API 1.0.0" style: + a STAC Item (type "Feature") + """ + + links: List[dict] = copy.deepcopy(result_metadata.links or job_info.links or []) + links = list_job_results_add_basic_links(links=links, job_id=job_id, user_id=user_id) + + assets = { + filename: _asset_object( + job_id=job_id, + user_id=user_id, + filename=filename, + asset_metadata=asset_metadata, + job_info=job_info, + stac11=False, + ) + for filename, asset_metadata in result_metadata.assets.items() + if asset_metadata.get("asset", True) + } + + result = { + "type": "Feature", + "stac_version": "1.0.0", + "id": job_info.id, + "properties": _properties_from_job_info(job_info), + "assets": assets, + "links": links, + "openeo:status": PARTIAL_JOB_STATUS.FINISHED, + } + if result_metadata.providers: + result["providers"] = result_metadata.providers + + geometry = job_info.geometry + result["geometry"] = geometry + if geometry: + result["bbox"] = job_info.bbox + + result["stac_extensions"] = [ + STAC_EXTENSION.PROCESSING, + STAC_EXTENSION.CARD4LOPTICAL, + STAC_EXTENSION.FILEINFO, + ] + + if sniff_stac_extension_prefix(result["assets"].values(), prefix="eo:"): + result["stac_extensions"].append(STAC_EXTENSION.EO_V110) + + if any(key.startswith("proj:") for key in result["properties"]) or any( + key.startswith("proj:") for key in result["assets"] + ): + result["stac_extensions"].append(STAC_EXTENSION.PROJECTION_V120) + + return result + + +def _asset_object( + job_id, user_id, filename: str, asset_metadata: dict, job_info: BatchJobMetadata, stac11: bool +) -> dict: + result_dict = dict_no_none( + { + "title": asset_metadata.get("title", filename), + "href": asset_metadata.get(BatchJobs.ASSET_PUBLIC_HREF) + or get_backend_config().asset_url.build_url( + asset_metadata=asset_metadata, asset_name=filename, job_id=job_id, user_id=user_id + ), + "type": asset_metadata.get("type", asset_metadata.get("media_type", "application/octet-stream")), + "roles": asset_metadata.get("roles", ["data"]), + # TODO: eliminate this legacy "raster:bands" construct at some point? + "raster:bands": None if stac11 else asset_metadata.get("raster:bands"), + "file:size": asset_metadata.get("file:size"), + "alternate": asset_metadata.get("alternate"), + } + ) + if filename.endswith(".model"): + # Machine learning models. + return result_dict + bands = asset_metadata.get("bands") + + if bands: + # TODO: #298 this is a quick stop-gap solution for lack of clear API + # what "bands" actually is expected to be: + # a list of Band objects (current approach in openeo-geopyspark-driver) + # or a list of dictionaries (as handled in openeo-aggregator) + # TODO: move this normalization to a more general utility? + bands = [ + openeo.metadata.Band( + name=b.get("name"), + common_name=b.get("eo:common_name") or b.get("common_name"), + wavelength_um=b.get("eo:center_wavelength") or b.get("center_wavelength"), + ) + if isinstance(b, dict) + else b + for b in bands + ] + + # TODO: eliminate this legacy "eo:bands" construct at some point? + if not stac11: + result_dict["eo:bands"] = [ + dict_no_none( + { + "name": band.name, + "common_name": band.common_name, + "center_wavelength": band.wavelength_um, + } + ) + for band in bands + ] + else: + + def raster_bands(band_index) -> dict: + rb = asset_metadata.get("raster:bands", []) + return rb[band_index] if band_index < len(rb) else {} + + result_dict["bands"] = [ + dict_no_none( + { + **{ + "name": band.name, + "eo:common_name": band.common_name, + "eo:center_wavelength": band.wavelength_um, + }, + **raster_bands(i), + } + ) + for (i, band) in enumerate(bands) + ] + + asset_proj_epsg = asset_metadata.get("proj:epsg", job_info.epsg) + result_dict.update( + dict_no_none( + { + "proj:bbox": asset_metadata.get("proj:bbox", job_info.proj_bbox), + "proj:epsg": asset_proj_epsg, + "proj:code": f"EPSG:{asset_proj_epsg}" if asset_proj_epsg else None, + "proj:shape": asset_metadata.get("proj:shape", job_info.proj_shape), + } + ) + ) + + if "file:size" not in result_dict and "output_dir" in asset_metadata: + the_file = Path(asset_metadata["output_dir"]) / filename + if the_file.exists(): + size_in_bytes = the_file.stat().st_size + result_dict["file:size"] = size_in_bytes + + return result_dict + + +def _properties_from_job_info(job_info: BatchJobMetadata) -> dict: + to_datetime = Rfc3339(propagate_none=True).datetime + + properties = dict_no_none( + { + "title": job_info.title, + "description": job_info.description, + "created": to_datetime(job_info.created), + "updated": to_datetime(job_info.updated), + "card4l:specification": "SR", + "card4l:specification_version": "5.0", + "processing:facility": get_backend_config().processing_facility, + "processing:software": get_backend_config().processing_software, + } + ) + properties["datetime"] = None + + start_datetime = to_datetime(job_info.start_datetime) + end_datetime = to_datetime(job_info.end_datetime) + + if start_datetime == end_datetime: + properties["datetime"] = start_datetime + else: + if start_datetime: + properties["start_datetime"] = start_datetime + if end_datetime: + properties["end_datetime"] = end_datetime + + if job_info.instruments: + properties["instruments"] = job_info.instruments + + if job_info.epsg: + properties["proj:epsg"] = job_info.epsg + properties["proj:code"] = f"EPSG:{job_info.epsg}" + + if job_info.proj_bbox: + properties["proj:bbox"] = job_info.proj_bbox + + if job_info.proj_shape: + properties["proj:shape"] = job_info.proj_shape + + properties["card4l:processing_chain"] = job_info.process + + return properties + + +def _normalize_job_result_link(link: dict, *, job_id: str, user_id: str) -> dict: + if link.get(ITEM_LINK_PROPERTY.EXPOSE_AUXILIARY, False): + link = _auxiliary_link(exposable_link=link, job_id=job_id, user_id=user_id) + + if link.get("rel") == "original": + # TODO: Cleanup + # TODO: this "original" handling is highly specific to a niche openeo-geopyspark-driver feature (CWL) + # and does not really fit the generic nature of openeo-python-driver. + # Can this be generalized more cleanly? Or moved to openeo-geopyspark-driver? + try: + # TODO: assumes file is not nested + asset_name = urllib.parse.urlparse(link["href"]).path.split("/")[-1] + href = flask.url_for( + ".download_job_result", + job_id=job_id, + filename=asset_name, + _external=True, + ) + link = dict(**link, href=href) + except Exception as e: + _log.warning("Error when making URL for 'original' link: " + str(e)) + + return link + + +def _auxiliary_link(exposable_link: dict, *, job_id: str, user_id: str) -> dict: + auxiliary_filename = urllib.parse.urlparse(exposable_link["href"]).path.split("/")[ + -1 + ] # TODO: assumes file is not nested + + if exposable_link["href"].startswith("s3://"): + # TODO: asset.build_url is made for assets, but not aux links, right? + href = get_backend_config().asset_url.build_url( + asset_metadata={"href": exposable_link["href"]}, # TODO: clean up this hack to support s3proxy + asset_name=auxiliary_filename, + job_id=job_id, + user_id=user_id, + ) + else: + signer = get_backend_config().url_signer + if signer: + expires = signer.get_expires() + secure_key = signer.sign_job_asset( + job_id=job_id, user_id=user_id, filename=auxiliary_filename, expires=expires + ) + user_base64 = user_id_b64_encode(user_id) + href = flask.url_for( + ".download_job_auxiliary_file_signed", + job_id=job_id, + user_base64=user_base64, + filename=auxiliary_filename, + expires=expires, + secure_key=secure_key, + _external=True, + ) + else: + href = flask.url_for( + ".download_job_auxiliary_file", job_id=job_id, filename=auxiliary_filename, _external=True + ) + + return dict_no_none( + href=href, + rel=exposable_link.get("rel"), + type=exposable_link.get("type"), + ) + + +def get_item_metadata_doc( + *, batch_jobs: BatchJobs, job_id: str, item_id: str, user_id: str, format: Optional[str] = None +) -> dict: + if format == "stac11": + return _get_job_result_item11(batch_jobs=batch_jobs, job_id=job_id, item_id=item_id, user_id=user_id) + else: + return _get_job_result_item(batch_jobs=batch_jobs, job_id=job_id, item_id=item_id, user_id=user_id) + + +def _get_job_result_item(*, batch_jobs: BatchJobs, job_id: str, item_id: str, user_id: str) -> dict: + if item_id == DriverMlModel.METADATA_FILE_NAME: + return _download_ml_model_metadata(batch_jobs=batch_jobs, job_id=job_id, file_name=item_id, user_id=user_id) + + results = batch_jobs.get_result_assets(job_id=job_id, user_id=user_id) + + assets_for_item_id = { + asset_filename: metadata for asset_filename, metadata in results.items() if asset_filename.startswith(item_id) + } + + if len(assets_for_item_id) != 1: + raise AssertionError(f"expected exactly 1 asset with file name {item_id}. Got {len(assets_for_item_id)}") + + asset_filename, asset_metadata = next(iter(assets_for_item_id.items())) + + job_info = batch_jobs.get_job_info(job_id, user_id) + + properties = {"datetime": asset_metadata.get("datetime")} + if properties["datetime"] is None: + to_datetime = Rfc3339(propagate_none=True).datetime + + start_datetime = asset_metadata.get("start_datetime") or to_datetime(job_info.start_datetime) + end_datetime = asset_metadata.get("end_datetime") or to_datetime(job_info.end_datetime) + + if start_datetime == end_datetime: + properties["datetime"] = start_datetime + else: + if start_datetime: + properties["start_datetime"] = start_datetime + if end_datetime: + properties["end_datetime"] = end_datetime + + if job_info.proj_shape: + properties["proj:shape"] = job_info.proj_shape + if job_info.proj_bbox: + properties["proj:bbox"] = job_info.proj_bbox + if job_info.epsg: + properties["proj:epsg"] = job_info.epsg + properties["proj:code"] = f"EPSG:{job_info.epsg}" + + bbox = asset_metadata.get("bbox", job_info.bbox) + if not bbox and job_info.proj_bbox and job_info.epsg: + bbox = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).reproject(4326).as_wsen_tuple() + geometry = asset_metadata.get("geometry", job_info.geometry) + if not geometry and job_info.proj_bbox and job_info.epsg: + geometry = BoundingBox.from_wsen_tuple(wsen=job_info.proj_bbox, crs=job_info.epsg).as_geojson() + + stac_item = { + "type": "Feature", + "stac_version": "1.0.0", + "stac_extensions": [ + STAC_EXTENSION.EO_V110, + STAC_EXTENSION.FILEINFO, + STAC_EXTENSION.PROJECTION_V120, + ], + "id": item_id, + "geometry": geometry, + "bbox": bbox, + "properties": properties, + "links": [ + { + "rel": "self", + # MUST be absolute + "href": flask.url_for(".get_job_result_item", job_id=job_id, item_id=item_id, _external=True), + "type": STAC_ITEM_MEDIA_TYPE, + }, + { + "rel": "collection", + "href": flask.url_for(".list_job_results", job_id=job_id, _external=True), # SHOULD be absolute + "type": "application/json", + }, + ], + "assets": { + asset_filename: _asset_object(job_id, user_id, asset_filename, asset_metadata, job_info, stac11=False) + }, + "collection": job_id, + } + # Add optional items, if they are present. + stac_item.update( + **dict_no_none( + { + "epsg": job_info.epsg, + } + ) + ) + return stac_item + + +def _get_job_result_item11(*, batch_jobs: BatchJobs, job_id, item_id, user_id) -> dict: + if item_id == DriverMlModel.METADATA_FILE_NAME: + return _download_ml_model_metadata(batch_jobs=batch_jobs, job_id=job_id, file_name=item_id, user_id=user_id) + + metadata = batch_jobs.get_result_metadata(job_id=job_id, user_id=user_id) + + if item_id not in metadata.items: + raise OpenEOApiException( + "Item with id {item_id!r} not found in job {job_id!r}".format(item_id=item_id, job_id=job_id), + status_code=404, + ) + item_metadata = metadata.items.get(item_id, None) + + job_info = batch_jobs.get_job_info(job_id, user_id) + + assets = {} + for asset_key, asset in item_metadata.get("assets", {}).items(): + if "output_dir" in asset: + out_dir = asset.get("output_dir") + _log.info(f"asset has output dir {out_dir} and href {asset.get('href')}") + common = os.path.commonpath([asset.get("href"), out_dir]) + href = os.path.relpath(asset.get("href"), common) + else: + _log.info(f"asset has no output dir and href {asset.get('href')}") + href = asset.get("href") + assets[asset_key] = _asset_object(job_id, user_id, href, asset, job_info, stac11=True) + + properties = item_metadata.get("properties", {"datetime": item_metadata.get("datetime")}) + if properties["datetime"] is None: + to_datetime = Rfc3339(propagate_none=True).datetime + + start_datetime = item_metadata.get("start_datetime") or to_datetime(job_info.start_datetime) + end_datetime = item_metadata.get("end_datetime") or to_datetime(job_info.end_datetime) + + if start_datetime == end_datetime: + properties["datetime"] = start_datetime + else: + if start_datetime: + properties["start_datetime"] = start_datetime + if end_datetime: + properties["end_datetime"] = end_datetime + + if job_info.proj_shape: + properties["proj:shape"] = job_info.proj_shape + if job_info.proj_bbox: + properties["proj:bbox"] = job_info.proj_bbox + if job_info.epsg: + properties["proj:epsg"] = job_info.epsg + properties["proj:code"] = f"EPSG:{job_info.epsg}" + + bbox = item_metadata.get("bbox", job_info.bbox) + if not bbox and job_info.proj_bbox and job_info.epsg: + bbox = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).reproject(4326).as_wsen_tuple() + geometry = item_metadata.get("geometry", job_info.geometry) + if not geometry and job_info.proj_bbox and job_info.epsg: + geometry = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).as_geojson() + + auxiliary_links = [ + _auxiliary_link(link, job_id=job_id, user_id=user_id) + for link in item_metadata.get("links", []) + if link.get(ITEM_LINK_PROPERTY.EXPOSE_AUXILIARY, False) + ] + + stac_item = { + "type": "Feature", + "stac_version": "1.1.0", + "stac_extensions": [ + STAC_EXTENSION.EO_V110, + STAC_EXTENSION.FILEINFO, + STAC_EXTENSION.PROJECTION_V120, + ], + "id": item_id, + "geometry": geometry, + "bbox": bbox, + "properties": properties, + "links": [ + { + "rel": "self", + # MUST be absolute + "href": flask.url_for(".get_job_result_item11", job_id=job_id, item_id=item_id, _external=True), + "type": STAC_ITEM_MEDIA_TYPE, + }, + { + "rel": "collection", + "href": flask.url_for(".list_job_results", job_id=job_id, _external=True), # SHOULD be absolute + "type": "application/json", + }, + ] + + auxiliary_links, + "assets": assets, + "collection": job_id, + } + # Add optional items, if they are present. + stac_item.update( + **dict_no_none( + { + "epsg": job_info.epsg, + } + ) + ) + return stac_item + + +def _download_ml_model_metadata(*, batch_jobs: BatchJobs, job_id: str, file_name: str, user_id) -> dict: + results = batch_jobs.get_result_assets(job_id=job_id, user_id=user_id) + ml_model_metadata: dict = results.get(file_name, None) + if ml_model_metadata is None: + raise FilePathInvalidException(f"{file_name!r} not in {list(results.keys())}") + assets = deep_get(ml_model_metadata, "assets", default={}) + for asset in assets.values(): + if not asset["href"].startswith("http"): + asset_file_name = Path(asset["href"]).name + asset["href"] = get_backend_config().asset_url.build_url( + asset_metadata=asset, asset_name=asset_file_name, job_id=job_id, user_id=user_id + ) + stac_item = { + "stac_version": ml_model_metadata.get("stac_version", "1.0.0"), + "stac_extensions": ml_model_metadata.get("stac_extensions", []), + "type": "Feature", + "id": ml_model_metadata.get("id"), + "collection": job_id, + "bbox": ml_model_metadata.get("bbox", []), + "geometry": ml_model_metadata.get("geometry", {}), + "properties": ml_model_metadata.get("properties", {}), + "links": ml_model_metadata.get("links", []), + "assets": ml_model_metadata.get("assets", {}), + } + return stac_item diff --git a/openeo_driver/constants.py b/openeo_driver/constants.py index 9d71cf13..a5aa08ab 100644 --- a/openeo_driver/constants.py +++ b/openeo_driver/constants.py @@ -82,3 +82,6 @@ class ITEM_LINK_PROPERTY: class LINK_REL: # Custom openEO link relation for OGC queryables, per https://github.com/Open-EO/openeo-api/pull/487 OGC_QUERYABLES = "http://www.opengis.net/def/rel/ogc/1.0/queryables" + + +STAC_ITEM_MEDIA_TYPE = "application/geo+json" diff --git a/openeo_driver/views.py b/openeo_driver/views.py index 5be351f6..5a3bb03c 100644 --- a/openeo_driver/views.py +++ b/openeo_driver/views.py @@ -14,6 +14,7 @@ import flask import flask_cors from flask import ( + # TODO: avoid import of these symbols with short/generic name, prefer `flask.` prefix usage Blueprint, Flask, abort, @@ -26,24 +27,20 @@ send_from_directory, url_for, ) -import openeo.metadata -from openeo.util import Rfc3339, TimingLogger, deep_get, dict_no_none, rfc3339 +from openeo.util import Rfc3339, TimingLogger, deep_get, dict_no_none from openeo.utils.version import ComparableVersion -from pyproj import CRS -import shapely.geometry from werkzeug.exceptions import HTTPException, NotFound from werkzeug.middleware.proxy_fix import ProxyFix from openeo_driver.backend import ( BatchJobMetadata, - BatchJobs, ErrorSummary, JobListing, OpenEoBackendImplementation, + QueryablesListing, ServiceMetadata, UserDefinedProcessMetadata, is_not_implemented, - QueryablesListing, ) from openeo_driver.config import OpenEoBackendConfig, get_backend_config from openeo_driver.constants import ( @@ -51,16 +48,14 @@ DEFAULT_LOG_LEVEL_RETRIEVAL, ITEM_LINK_PROPERTY, JOB_STATUS, - STAC_EXTENSION, LINK_REL, + STAC_EXTENSION, + STAC_ITEM_MEDIA_TYPE, ) -from openeo_driver.datacube import DriverMlModel from openeo_driver.dry_run import DryRunDataTracer from openeo_driver.errors import ( FeatureUnsupportedException, - FilePathInvalidException, InternalException, - JobNotFinishedException, NotFoundException, OpenEOApiException, ProcessGraphInvalidException, @@ -69,19 +64,15 @@ ProcessUnsupportedException, ServiceNotFoundException, ) -from openeo_driver.jobregistry import PARTIAL_JOB_STATUS from openeo_driver.processgraph import ProcessGraphFlatDict, extract_default_job_options_from_process_graph from openeo_driver.save_result import SaveResult, to_save_result -from openeo_driver.users import User, user_id_b64_decode, user_id_b64_encode +from openeo_driver.users import User, user_id_b64_decode from openeo_driver.users.auth import HttpAuthHandler -from openeo_driver.util.compat import function_has_argument, filter_supported_kwargs -from openeo_driver.util.geometry import BoundingBox, reproject_geometry +from openeo_driver.util.compat import filter_supported_kwargs, function_has_argument from openeo_driver.util.logging import ExtraLoggingFilter, FlaskRequestCorrelationIdLogging -from openeo_driver.util.stac import sniff_stac_extension_prefix from openeo_driver.util.stac_utils import get_files_from_stac_catalog from openeo_driver.utils import EvalEnv, smart_bool - _log = logging.getLogger(__name__) ApiVersionInfo = namedtuple("ApiVersionInfo", ["version", "supported", "wellknown", "production"]) @@ -857,52 +848,6 @@ def processes_details(namespace, process_id): return jsonify(process) -def _properties_from_job_info(job_info: BatchJobMetadata) -> dict: - to_datetime = Rfc3339(propagate_none=True).datetime - - properties = dict_no_none( - { - "title": job_info.title, - "description": job_info.description, - "created": to_datetime(job_info.created), - "updated": to_datetime(job_info.updated), - "card4l:specification": "SR", - "card4l:specification_version": "5.0", - "processing:facility": get_backend_config().processing_facility, - "processing:software": get_backend_config().processing_software, - } - ) - properties["datetime"] = None - - start_datetime = to_datetime(job_info.start_datetime) - end_datetime = to_datetime(job_info.end_datetime) - - if start_datetime == end_datetime: - properties["datetime"] = start_datetime - else: - if start_datetime: - properties["start_datetime"] = start_datetime - if end_datetime: - properties["end_datetime"] = end_datetime - - if job_info.instruments: - properties["instruments"] = job_info.instruments - - if job_info.epsg: - properties["proj:epsg"] = job_info.epsg - properties["proj:code"] = f"EPSG:{job_info.epsg}" - - if job_info.proj_bbox: - properties["proj:bbox"] = job_info.proj_bbox - - if job_info.proj_shape: - properties["proj:shape"] = job_info.proj_shape - - properties["card4l:processing_chain"] = job_info.process - - return properties - - def _assert_valid_log_level(level: str) -> str: valid_levels = ["debug", "info", "warning", "error"] if level not in valid_levels: @@ -918,7 +863,6 @@ def register_views_batch_jobs( blueprint: Blueprint, backend_implementation: OpenEoBackendImplementation, api_endpoint: EndpointRegistry, auth_handler: HttpAuthHandler ): - stac_item_media_type = "application/geo+json" @api_endpoint @blueprint.route('/jobs', methods=['POST']) @@ -1049,9 +993,13 @@ def start_job(job_id, user: User): def list_job_results(job_id, user: User): # TODO: How is "partial" encoded? Also see https://github.com/Open-EO/openeo-api/issues/509 partial = str(request.args.get("partial")).lower() in {"true", "1"} - return _list_job_results(job_id, user.user_id, partial=partial) - @blueprint.route('/jobs//results//', methods=['GET']) + doc = backend_implementation.batch_jobs.list_job_results( + job_id=job_id, user_id=user.user_id, partial=partial, api_version=requested_api_version() + ) + return flask.jsonify(doc) + + @blueprint.route("/jobs//results//", methods=["GET"]) def list_job_results_signed(job_id, user_base64, secure_key): expires = request.args.get('expires') signer = get_backend_config().url_signer @@ -1060,325 +1008,10 @@ def list_job_results_signed(job_id, user_base64, secure_key): # TODO: How is "partial" encoded? Also see https://github.com/Open-EO/openeo-api/issues/509 partial = str(request.args.get("partial")).lower() in {"true", "1"} with TimingLogger(f"_list_job_results({job_id=}, {user_id=}, {partial=})", _log): - return _list_job_results(job_id, user_id, partial=partial) - - def _list_job_results(job_id, user_id, *, partial: bool = False): - to_datetime = Rfc3339(propagate_none=True).datetime - - def job_results_canonical_url() -> str: - signer = get_backend_config().url_signer - if not signer: - return url_for(".list_job_results", job_id=job_id, _external=True) - - expires = signer.get_expires() - secure_key = signer.sign_job_results(job_id=job_id, user_id=user_id, expires=expires) - user_base64 = user_id_b64_encode(user_id) - # TODO: also encrypt user id? - # TODO: encode all stuff (signature, userid, expiry) in a single blob in the URL - - if partial: - return url_for( - ".list_job_results_signed", - job_id=job_id, - user_base64=user_base64, - expires=expires, - secure_key=secure_key, - _external=True, - partial="true", - ) - else: - return url_for( - ".list_job_results_signed", - job_id=job_id, - user_base64=user_base64, - expires=expires, - secure_key=secure_key, - _external=True, - ) - - with TimingLogger(f"backend_implementation.batch_jobs.get_job_info({job_id=}, {user_id=})", _log): - job_info = backend_implementation.batch_jobs.get_job_info(job_id, user_id) - - if job_info.status != JOB_STATUS.FINISHED: - if not partial: - raise JobNotFinishedException() - else: - result = { - "openeo:status": PARTIAL_JOB_STATUS.for_job_status(job_info.status), - "type": "Collection", - "stac_version": "1.0.0", - "id": job_id, - "title": job_info.title or "Unfinished batch job {job_id}", - "description": job_info.description or f"Results for batch job {job_id}", - "license": "proprietary", # TODO? - "extent": { - "spatial": {"bbox": [[-180, -90, 180, 90]]}, - "temporal": {"interval": [[rfc3339.now_utc(), rfc3339.now_utc()]]}, - }, - "links": [ - { - "rel": "canonical", - "href": job_results_canonical_url(), - "type": "application/json", - } - ], - } - return jsonify(result) - - with TimingLogger(f"backend_implementation.batch_jobs.get_result_metadata({job_id=}, {user_id=})", _log): - result_metadata = backend_implementation.batch_jobs.get_result_metadata( - job_id=job_id, user_id=user_id + doc = backend_implementation.batch_jobs.list_job_results( + job_id=job_id, user_id=user_id, partial=partial, api_version=requested_api_version() ) - result_assets = result_metadata.assets - result_items = result_metadata.items - providers = result_metadata.providers - - links: List[dict] = copy.deepcopy(result_metadata.links or job_info.links or []) - - try: - # TODO: Cleanup - original_link = next((l for l in links if l.get("rel") == "original"), None) - if original_link: - asset_name = original_link["href"][original_link["href"].rindex("/") + 1 :] - original_link["href"] = url_for( - ".download_job_result", - job_id=job_id, - filename=asset_name, - _external=True, - ) - except Exception as e: - _log.warning("Error when making URL for 'original' link: " + str(e)) - - if not any(l.get("rel") == "self" for l in links): - links.append( - { - "rel": "self", - "href": url_for(".list_job_results", job_id=job_id, _external=True), # MUST be absolute - "type": "application/json", - } - ) - if not any(l.get("rel") == "canonical" for l in links): - links.append( - { - "rel": "canonical", - "href": job_results_canonical_url(), - "type": "application/json", - } - ) - if not any(l.get("rel") == "card4l-document" for l in links): - links.append( - { - "rel": "card4l-document", - # TODO: avoid hardcoding this specific URL? - "href": "http://ceos.org/ard/files/PFS/SR/v5.0/CARD4L_Product_Family_Specification_Surface_Reflectance-v5.0.pdf", - "type": "application/pdf", - } - ) - - assets = { - filename: _asset_object( - job_id=job_id, - user_id=user_id, - filename=filename, - asset_metadata=asset_metadata, - job_info=job_info, - stac11=False, - ) - for filename, asset_metadata in result_assets.items() - if asset_metadata.get("asset", True) - } - - if requested_api_version().at_least("1.1.0"): - ml_model_metadata = None - - def job_result_item_url(item_id, is11 = False) -> str: - signer = get_backend_config().url_signer - - method_start = ".get_job_result_item" - if is11: - method_start = method_start + "11" - if not signer: - return url_for(method_start, job_id=job_id, item_id=item_id, _external=True) - - expires = signer.get_expires() - secure_key = signer.sign_job_item(job_id=job_id, user_id=user_id, item_id=item_id, expires=expires) - user_base64 = user_id_b64_encode(user_id) - return url_for( - method_start + "_signed", - job_id=job_id, - user_base64=user_base64, - secure_key=secure_key, - item_id=item_id, - expires=expires, - _external=True, - ) - - - if result_metadata.items : - def intersect_band_array(list1, list2): - band_result = [] - for item1 in list1: - if isinstance(item1, dict) and "name" in item1: - for item2 in list2: - if isinstance(item1, dict) and "name" in item1 and item1["name"] == item2["name"]: - band_result.append(intersect_dicts(item1, item2)) - return band_result - - def intersect_dicts(dict1, dict2): - result = {} - for key in dict1: - if key in dict2: - if isinstance(dict1[key], dict) and isinstance(dict2[key], dict): - # Recursively intersect nested dictionaries - nested_result = intersect_dicts(dict1[key], dict2[key]) - if nested_result: # Only add if the nested result is not empty - result[key] = nested_result - elif isinstance(dict1[key], list) and isinstance(dict2[key], list) and key == "bands": - result[key] = intersect_band_array(dict1[key], dict2[key]) - elif dict1[key] == dict2[key]: - # Retain the key-value pair if values are equal - result[key] = dict1[key] - return result - - item_assets = {} - assets = {} - for item_key, item_metadata in result_items.items(): - for asset_key, asset_metadata in item_metadata.get("assets", {}).items(): - if "output_dir" in asset_metadata: - out_dir = asset_metadata.get("output_dir") - _log.info(f"asset has output dir {out_dir} and href {asset_metadata.get('href')}") - common = os.path.commonpath([asset_metadata.get('href'), out_dir]) - href = os.path.relpath(asset_metadata.get('href'),common) - else: - href = asset_metadata.get("href") - asset_object = _asset_object( - job_id=job_id, - user_id=user_id, - filename= href, - asset_metadata=asset_metadata, - job_info=job_info, - stac11=True, - ) - assets[item_key + "_" + asset_key] = asset_object - item_asset = dict_no_none( - { - "type": asset_object.get("type"), - "roles": asset_object.get("roles"), - "bands": asset_object.get("bands"), - "proj:bbox": asset_object.get("proj:bbox"), - "proj:epsg": asset_object.get("proj:epsg"), - "proj:shape": asset_object.get("proj:shape"), - "file:size": asset_object.get("file:size"), - } - ) - if asset_key not in item_assets: - item_assets[asset_key] = item_asset - else: - item_assets[asset_key] = intersect_dicts(item_assets[asset_key], item_asset) - for item_id in result_metadata.items.keys(): - links.append( - {"rel": "item", "href": job_result_item_url(item_id=item_id, is11=True), "type": stac_item_media_type} - ) - stac_version = "1.1.0" - else: - - item_assets = None - for filename, metadata in result_assets.items(): - if ("data" in metadata.get("roles", []) and - any(media_type in metadata.get("type", "") for media_type in - ["geotiff", "netcdf", "text/csv", "application/parquet"])): - links.append( - {"rel": "item", "href": job_result_item_url(item_id=filename), "type": stac_item_media_type} - ) - elif metadata.get("ml_model_metadata", False): - # TODO: Currently we only support one ml_model per batch job. - ml_model_metadata = metadata - links.append( - {"rel": "item", "href": job_result_item_url(item_id=filename), "type": "application/json"} - ) - stac_version = "1.0.0" - - links = [_normalize_job_result_link(link=k, job_id=job_id, user_id=user_id) for k in links] - - result = dict_no_none( - { - "type": "Collection", - "stac_version": stac_version, - "stac_extensions": [ - STAC_EXTENSION.EO_V110, - STAC_EXTENSION.FILEINFO, - STAC_EXTENSION.PROCESSING, - STAC_EXTENSION.PROJECTION_V120, - ], - "id": job_id, - "title": job_info.title, - "description": job_info.description or f"Results for batch job {job_id}", - "license": "proprietary", # TODO? - "extent": { - "spatial": {"bbox": [job_info.bbox] if job_info.bbox else [[-180, -90, 180, 90]]}, - "temporal": { - "interval": [[to_datetime(job_info.start_datetime), to_datetime(job_info.end_datetime)]] - }, - }, - "summaries": {"instruments": job_info.instruments } if job_info.instruments else {}, - "providers": providers or None, - "links": links, - "assets": assets, - "item_assets": item_assets, - "openeo:status": PARTIAL_JOB_STATUS.FINISHED, - } - ) - - if ml_model_metadata is not None: - result["stac_extensions"].extend(ml_model_metadata.get("stac_extensions", [])) - if "summaries" not in result.keys(): - result["summaries"] = {} - if "properties" in ml_model_metadata.keys(): - ml_model_properties = ml_model_metadata["properties"] - learning_approach = ml_model_properties.get("ml-model:learning_approach", None) - prediction_type = ml_model_properties.get("ml-model:prediction_type", None) - architecture = ml_model_properties.get("ml-model:architecture", None) - result["summaries"].update( - { - "ml-model:learning_approach": [learning_approach] if learning_approach is not None else [], - "ml-model:prediction_type": [prediction_type] if prediction_type is not None else [], - "ml-model:architecture": [architecture] if architecture is not None else [], - } - ) - else: - result = { - "type": "Feature", - "stac_version": "1.0.0", - "id": job_info.id, - "properties": _properties_from_job_info(job_info), - "assets": assets, - "links": links, - "openeo:status": PARTIAL_JOB_STATUS.FINISHED, - } - if providers: - result["providers"] = providers - - geometry = job_info.geometry - result["geometry"] = geometry - if geometry: - result["bbox"] = job_info.bbox - - result["stac_extensions"] = [ - STAC_EXTENSION.PROCESSING, - STAC_EXTENSION.CARD4LOPTICAL, - STAC_EXTENSION.FILEINFO, - ] - - if sniff_stac_extension_prefix(result["assets"].values(), prefix="eo:"): - result["stac_extensions"].append(STAC_EXTENSION.EO_V110) - - if any(key.startswith("proj:") for key in result["properties"]) or any( - key.startswith("proj:") for key in result["assets"] - ): - result["stac_extensions"].append(STAC_EXTENSION.PROJECTION_V120) - - # TODO "OpenEO-Costs" header? - return jsonify(result) + return flask.jsonify(doc) # TODO: Issue #232, TBD: refactor download functionality? more abstract, just stream blocks of bytes from S3 or from a directory. def _download_job_result( @@ -1470,6 +1103,7 @@ def _download_job_result( def _stream_from_s3(s3_url, *, filename, mimetype: Optional[str], bytes_range: Optional[str]): # Local imports as S3 requirements are optional import botocore.exceptions + from openeo_driver.integrations.s3.client import S3ClientBuilder bucket, key = s3_url[5:].split("/", 1) @@ -1506,7 +1140,12 @@ def get_job_result_item_signed(job_id, user_base64, secure_key, item_id): signer = get_backend_config().url_signer user_id = user_id_b64_decode(user_base64) signer.verify_job_item(signature=secure_key, job_id=job_id, user_id=user_id, item_id=item_id, expires=expires) - return _get_job_result_item(job_id, item_id, user_id) + doc = backend_implementation.batch_jobs.get_item_metadata_doc( + job_id=job_id, user_id=user_id, item_id=item_id, format="openeo100" + ) + resp = flask.jsonify(doc) + resp.mimetype = STAC_ITEM_MEDIA_TYPE + return resp @api_endpoint @blueprint.route("/jobs//results/items11///", methods=["GET"]) @@ -1515,168 +1154,35 @@ def get_job_result_item11_signed(job_id, user_base64, secure_key, item_id): signer = get_backend_config().url_signer user_id = user_id_b64_decode(user_base64) signer.verify_job_item(signature=secure_key, job_id=job_id, user_id=user_id, item_id=item_id, expires=expires) - return _get_job_result_item11(job_id, item_id, user_id) + doc = backend_implementation.batch_jobs.get_item_metadata_doc( + job_id=job_id, user_id=user_id, item_id=item_id, format="stac11" + ) + resp = flask.jsonify(doc) + resp.mimetype = STAC_ITEM_MEDIA_TYPE + return resp - @blueprint.route('/jobs//results/items/', methods=['GET']) + @blueprint.route("/jobs//results/items/", methods=["GET"]) @auth_handler.requires_bearer_auth def get_job_result_item(job_id: str, item_id: str, user: User) -> flask.Response: - return _get_job_result_item(job_id, item_id, user.user_id) + doc = backend_implementation.batch_jobs.get_item_metadata_doc( + job_id=job_id, user_id=user.user_id, item_id=item_id, format="openeo100" + ) + resp = flask.jsonify(doc) + resp.mimetype = STAC_ITEM_MEDIA_TYPE + return resp + @api_endpoint(version=ComparableVersion("1.1.0").or_higher) @blueprint.route('/jobs//results/items11/', methods=['GET']) @auth_handler.requires_bearer_auth def get_job_result_item11(job_id: str, item_id: str, user: User) -> flask.Response: - return _get_job_result_item11(job_id, item_id, user.user_id) - - def _get_job_result_item11(job_id, item_id, user_id): - if item_id == DriverMlModel.METADATA_FILE_NAME: - return _download_ml_model_metadata(job_id, item_id, user_id) - - metadata = backend_implementation.batch_jobs.get_result_metadata( - job_id=job_id, user_id=user_id - ) - - if item_id not in metadata.items: - raise OpenEOApiException("Item with id {item_id!r} not found in job {job_id!r}".format(item_id=item_id, job_id=job_id), status_code=404) - item_metadata = metadata.items.get(item_id,None) - - job_info = backend_implementation.batch_jobs.get_job_info(job_id, user_id) - - assets = {} - for asset_key, asset in item_metadata.get("assets", {}).items(): - if "output_dir" in asset: - out_dir = asset.get("output_dir") - _log.info(f"asset has output dir {out_dir} and href {asset.get('href')}") - common = os.path.commonpath([asset.get('href'), out_dir]) - href = os.path.relpath(asset.get('href'), common) - else: - _log.info(f"asset has no output dir and href {asset.get('href')}") - href = asset.get("href") - assets[asset_key] = _asset_object(job_id, user_id, href, asset, job_info, stac11=True) - - - properties = item_metadata.get("properties", {"datetime": item_metadata.get("datetime")}) - if properties["datetime"] is None: - to_datetime = Rfc3339(propagate_none=True).datetime - - start_datetime = item_metadata.get("start_datetime") or to_datetime(job_info.start_datetime) - end_datetime = item_metadata.get("end_datetime") or to_datetime(job_info.end_datetime) - - if start_datetime == end_datetime: - properties["datetime"] = start_datetime - else: - if start_datetime: - properties["start_datetime"] = start_datetime - if end_datetime: - properties["end_datetime"] = end_datetime - - if job_info.proj_shape: - properties["proj:shape"] = job_info.proj_shape - if job_info.proj_bbox: - properties["proj:bbox"] = job_info.proj_bbox - if job_info.epsg: - properties["proj:epsg"] = job_info.epsg - properties["proj:code"] = f"EPSG:{job_info.epsg}" - - bbox = item_metadata.get("bbox", job_info.bbox) - if not bbox and job_info.proj_bbox and job_info.epsg: - bbox = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).reproject(4326).as_wsen_tuple() - geometry = item_metadata.get("geometry", job_info.geometry) - if not geometry and job_info.proj_bbox and job_info.epsg: - geometry = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).as_geojson() - - auxiliary_links = [ - _auxiliary_link(link, job_id=job_id, user_id=user_id) - for link in item_metadata.get("links", []) - if link.get(ITEM_LINK_PROPERTY.EXPOSE_AUXILIARY, False) - ] - - stac_item = { - "type": "Feature", - "stac_version": "1.1.0", - "stac_extensions": [ - STAC_EXTENSION.EO_V110, - STAC_EXTENSION.FILEINFO, - STAC_EXTENSION.PROJECTION_V120, - ], - "id": item_id, - "geometry": geometry, - "bbox": bbox, - "properties": properties, - "links": [ - { - "rel": "self", - # MUST be absolute - "href": url_for(".get_job_result_item11", job_id=job_id, item_id=item_id, _external=True), - "type": stac_item_media_type, - }, - { - "rel": "collection", - "href": url_for(".list_job_results", job_id=job_id, _external=True), # SHOULD be absolute - "type": "application/json", - }, - ] - + auxiliary_links, - "assets": assets, - "collection": job_id, - } - # Add optional items, if they are present. - stac_item.update( - **dict_no_none( - { - "epsg": job_info.epsg, - } - ) + doc = backend_implementation.batch_jobs.get_item_metadata_doc( + job_id=job_id, user_id=user.user_id, item_id=item_id, format="stac11" ) - - resp = jsonify(stac_item) - resp.mimetype = stac_item_media_type + resp = flask.jsonify(doc) + resp.mimetype = STAC_ITEM_MEDIA_TYPE return resp - def _auxiliary_link(exposable_link: dict, *, job_id: str, user_id: str) -> dict: - auxiliary_filename = urlparse(exposable_link["href"]).path.split("/")[-1] # TODO: assumes file is not nested - - if exposable_link["href"].startswith("s3://"): - # TODO: asset.build_url is made for assets, but not aux links, right? - href = backend_implementation.config.asset_url.build_url( - asset_metadata={"href": exposable_link["href"]}, # TODO: clean up this hack to support s3proxy - asset_name=auxiliary_filename, - job_id=job_id, - user_id=user_id, - ) - else: - signer = get_backend_config().url_signer - if signer: - expires = signer.get_expires() - secure_key = signer.sign_job_asset( - job_id=job_id, user_id=user_id, filename=auxiliary_filename, expires=expires - ) - user_base64 = user_id_b64_encode(user_id) - href = flask.url_for( - ".download_job_auxiliary_file_signed", - job_id=job_id, - user_base64=user_base64, - filename=auxiliary_filename, - expires=expires, - secure_key=secure_key, - _external=True, - ) - else: - href = flask.url_for( - ".download_job_auxiliary_file", job_id=job_id, filename=auxiliary_filename, _external=True - ) - - return dict_no_none( - href=href, - rel=exposable_link.get("rel"), - type=exposable_link.get("type"), - ) - - def _normalize_job_result_link(link: dict, *, job_id: str, user_id: str) -> dict: - if link.get(ITEM_LINK_PROPERTY.EXPOSE_AUXILIARY, False): - link = _auxiliary_link(exposable_link=link, job_id=job_id, user_id=user_id) - - return link @blueprint.route("/jobs//results/aux///", methods=["GET"]) def download_job_auxiliary_file_signed(job_id, user_base64, secure_key, filename): @@ -1723,219 +1229,6 @@ def _download_job_auxiliary_file(job_id, filename, user_id): auxiliary_file = pathlib.Path(uri_parts.path) return send_from_directory(auxiliary_file.parent, auxiliary_file.name, mimetype=auxiliary_type) - def _get_job_result_item(job_id, item_id, user_id): - if item_id == DriverMlModel.METADATA_FILE_NAME: - return _download_ml_model_metadata(job_id, item_id, user_id) - - results = backend_implementation.batch_jobs.get_result_assets( - job_id=job_id, user_id=user_id - ) - - assets_for_item_id = { - asset_filename: metadata for asset_filename, metadata in results.items() - if asset_filename.startswith(item_id) - } - - if len(assets_for_item_id) != 1: - raise AssertionError(f"expected exactly 1 asset with file name {item_id}. Got {len(assets_for_item_id)}") - - asset_filename, asset_metadata = next(iter(assets_for_item_id.items())) - - job_info = backend_implementation.batch_jobs.get_job_info(job_id, user_id) - - - properties = {"datetime": asset_metadata.get("datetime")} - if properties["datetime"] is None: - to_datetime = Rfc3339(propagate_none=True).datetime - - start_datetime = asset_metadata.get("start_datetime") or to_datetime(job_info.start_datetime) - end_datetime = asset_metadata.get("end_datetime") or to_datetime(job_info.end_datetime) - - if start_datetime == end_datetime: - properties["datetime"] = start_datetime - else: - if start_datetime: - properties["start_datetime"] = start_datetime - if end_datetime: - properties["end_datetime"] = end_datetime - - if job_info.proj_shape: - properties["proj:shape"] = job_info.proj_shape - if job_info.proj_bbox: - properties["proj:bbox"] = job_info.proj_bbox - if job_info.epsg: - properties["proj:epsg"] = job_info.epsg - properties["proj:code"] = f"EPSG:{job_info.epsg}" - - bbox = asset_metadata.get("bbox", job_info.bbox) - if not bbox and job_info.proj_bbox and job_info.epsg: - bbox = BoundingBox.from_wsen_tuple(job_info.proj_bbox, crs=job_info.epsg).reproject(4326).as_wsen_tuple() - geometry = asset_metadata.get("geometry", job_info.geometry) - if not geometry and job_info.proj_bbox and job_info.epsg: - geometry = BoundingBox.from_wsen_tuple(wsen=job_info.proj_bbox, crs=job_info.epsg).as_geojson() - - stac_item = { - "type": "Feature", - "stac_version": "1.0.0", - "stac_extensions": [ - STAC_EXTENSION.EO_V110, - STAC_EXTENSION.FILEINFO, - STAC_EXTENSION.PROJECTION_V120, - ], - "id": item_id, - "geometry": geometry, - "bbox": bbox, - "properties": properties, - "links": [ - { - "rel": "self", - # MUST be absolute - "href": url_for(".get_job_result_item", job_id=job_id, item_id=item_id, _external=True), - "type": stac_item_media_type, - }, - { - "rel": "collection", - "href": url_for(".list_job_results", job_id=job_id, _external=True), # SHOULD be absolute - "type": "application/json", - }, - ], - "assets": { - asset_filename: _asset_object(job_id, user_id, asset_filename, asset_metadata, job_info, stac11=False) - }, - "collection": job_id, - } - # Add optional items, if they are present. - stac_item.update( - **dict_no_none( - { - "epsg": job_info.epsg, - } - ) - ) - - resp = jsonify(stac_item) - resp.mimetype = stac_item_media_type - return resp - - def _download_ml_model_metadata(job_id: str, file_name: str, user_id) -> flask.Response: - results = backend_implementation.batch_jobs.get_result_assets(job_id=job_id, user_id=user_id) - ml_model_metadata: dict = results.get(file_name, None) - if ml_model_metadata is None: - raise FilePathInvalidException(f"{file_name!r} not in {list(results.keys())}") - assets = deep_get(ml_model_metadata, "assets", default={}) - for asset in assets.values(): - if not asset["href"].startswith("http"): - asset_file_name = pathlib.Path(asset["href"]).name - asset["href"] = backend_implementation.config.asset_url.build_url( - asset_metadata=asset, asset_name=asset_file_name, job_id=job_id, user_id=user_id - ) - stac_item = { - "stac_version": ml_model_metadata.get("stac_version", "1.0.0"), - "stac_extensions": ml_model_metadata.get("stac_extensions", []), - "type": "Feature", - "id": ml_model_metadata.get("id"), - "collection": job_id, - "bbox": ml_model_metadata.get("bbox", []), - "geometry": ml_model_metadata.get("geometry", {}), - 'properties': ml_model_metadata.get("properties", {}), - 'links': ml_model_metadata.get("links", []), - 'assets': ml_model_metadata.get("assets", {}) - } - resp = jsonify(stac_item) - resp.mimetype = stac_item_media_type - return resp - - def _asset_object( - job_id, user_id, filename: str, asset_metadata: dict, job_info: BatchJobMetadata, stac11: bool - ) -> dict: - result_dict = dict_no_none( - { - "title": asset_metadata.get("title", filename), - "href": asset_metadata.get(BatchJobs.ASSET_PUBLIC_HREF) - or backend_implementation.config.asset_url.build_url( - asset_metadata=asset_metadata, asset_name=filename, job_id=job_id, user_id=user_id - ), - "type": asset_metadata.get("type", asset_metadata.get("media_type", "application/octet-stream")), - "roles": asset_metadata.get("roles", ["data"]), - # TODO: eliminate this legacy "raster:bands" construct at some point? - "raster:bands": None if stac11 else asset_metadata.get("raster:bands"), - "file:size": asset_metadata.get("file:size"), - "alternate": asset_metadata.get("alternate"), - } - ) - if filename.endswith(".model"): - # Machine learning models. - return result_dict - bands = asset_metadata.get("bands") - - if bands: - # TODO: #298 this is a quick stop-gap solution for lack of clear API - # what "bands" actually is expected to be: - # a list of Band objects (current approach in openeo-geopyspark-driver) - # or a list of dictionaries (as handled in openeo-aggregator) - # TODO: move this normalization to a more general utility? - bands = [ - openeo.metadata.Band( - name=b.get("name"), - common_name=b.get("eo:common_name") or b.get("common_name"), - wavelength_um=b.get("eo:center_wavelength") or b.get("center_wavelength"), - ) - if isinstance(b, dict) - else b - for b in bands - ] - - # TODO: eliminate this legacy "eo:bands" construct at some point? - if not stac11: - result_dict["eo:bands"] = [ - dict_no_none( - { - "name": band.name, - "common_name": band.common_name, - "center_wavelength": band.wavelength_um, - } - ) - for band in bands - ] - else: - def raster_bands(band_index) -> dict: - rb = asset_metadata.get("raster:bands", []) - return rb[band_index] if band_index < len(rb) else {} - - result_dict["bands"] = [ - dict_no_none( - { - **{ - "name": band.name, - "eo:common_name": band.common_name, - "eo:center_wavelength": band.wavelength_um, - }, - **raster_bands(i), - } - ) - for (i, band) in enumerate(bands) - ] - - asset_proj_epsg = asset_metadata.get("proj:epsg", job_info.epsg) - result_dict.update( - dict_no_none( - { - "proj:bbox": asset_metadata.get("proj:bbox", job_info.proj_bbox), - "proj:epsg": asset_proj_epsg, - "proj:code": f"EPSG:{asset_proj_epsg}" if asset_proj_epsg else None, - "proj:shape": asset_metadata.get("proj:shape", job_info.proj_shape), - } - ) - ) - - if "file:size" not in result_dict and "output_dir" in asset_metadata: - the_file = pathlib.Path(asset_metadata["output_dir"]) / filename - if the_file.exists(): - size_in_bytes = the_file.stat().st_size - result_dict["file:size"] = size_in_bytes - - return result_dict - @blueprint.route("/jobs//results/assets/", methods=["GET"]) @auth_handler.requires_bearer_auth def download_job_result(job_id, filename, user: User): diff --git a/openeo_driver/views_/__init__.py b/openeo_driver/views_/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openeo_driver/views_/batch_jobs.py b/openeo_driver/views_/batch_jobs.py new file mode 100644 index 00000000..e1ba3f1a --- /dev/null +++ b/openeo_driver/views_/batch_jobs.py @@ -0,0 +1,112 @@ +from typing import List, Union, Literal, Iterable + +import flask + +from openeo_driver.config import get_backend_config +from openeo_driver.users import user_id_b64_encode + + +def list_job_results_self_link(job_id: str) -> dict: + params = {k: v for k, v in flask.request.args.items() if k in {"partial"}} + url = flask.url_for(".list_job_results", job_id=job_id, _external=True, **params) + return { + "rel": "self", + "href": url, + # TODO: mime type should probably be "application/geo+json", + # but changing that might change too much tests to be feasible for now + "type": "application/json", + } + + +def list_job_results_canonical_link(job_id: str, *, user_id: str, partial: Union[bool, None] = None) -> dict: + extra = {} + if partial is not None: + extra["partial"] = str(partial).lower() + + signer = get_backend_config().url_signer + if signer: + expires = signer.get_expires() + secure_key = signer.sign_job_results(job_id=job_id, user_id=user_id, expires=expires) + user_base64 = user_id_b64_encode(user_id) + url = flask.url_for( + ".list_job_results_signed", + job_id=job_id, + user_base64=user_base64, + expires=expires, + secure_key=secure_key, + _external=True, + **extra, + ) + else: + url = flask.url_for(".list_job_results", job_id=job_id, _external=True, **extra) + + return { + "rel": "canonical", + "href": url, + # TODO: mime type should probably be "application/geo+json", + # but changing that might change too much tests to be feasible for now + "type": "application/json", + } + + +def card4l_link() -> dict: + return { + "rel": "card4l-document", + # TODO: avoid hardcoding this specific URL? + "href": "http://ceos.org/ard/files/PFS/SR/v5.0/CARD4L_Product_Family_Specification_Surface_Reflectance-v5.0.pdf", + "type": "application/pdf", + } + + +def list_job_results_add_basic_links( + links: List[dict], + *, + job_id: str, + user_id: str, + partial: Union[bool, None] = None, + add_self: bool = True, + add_canonical: bool = True, + add_card4l: bool = True, +) -> List[dict]: + """ + Add basic (self, canonical, ...) links in-place to the given links list + """ + if add_self: + links = add_link_by_rel(links, link=list_job_results_self_link(job_id=job_id), mode="fallback") + if add_canonical: + links = add_link_by_rel( + links, + link=list_job_results_canonical_link(job_id=job_id, user_id=user_id, partial=partial), + mode="fallback", + ) + if add_card4l: + links = add_link_by_rel(links, link=card4l_link(), mode="fallback") + + return links + + +def add_link_by_rel( + links: Iterable[dict], *, link: dict, mode: Literal["append", "fallback", "overwrite"] = "append" +) -> List[dict]: + """ + Add a link to the given collection of links, producing a new list links, + taking care of the "rel" attribute, e.g. to avoid duplicates: + - "append": always append the new link to the list + - "fallback": only append the new link if no link with the same rel already exists + - "overwrite": remove any existing links with the same rel before appending the new link + """ + # TODO: move this utility to a more generic place + + # Work on a copy + links = list(links) + + if mode == "append": + links += [link] + elif mode == "fallback": + if not any(l.get("rel") == link.get("rel") for l in links): + links.append(link) + elif mode == "overwrite": + links = [l for l in links if l.get("rel") != link.get("rel")] + [link] + else: + raise ValueError(f"Invalid {mode=}") + return links diff --git a/tests/test_views.py b/tests/test_views.py index 0729eddd..afe52c53 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -1861,22 +1861,28 @@ def test_get_job_results_unfinished_with_partial_true(self, api, job_status, exp resp.assert_status_code(200) job_result = resp.json - expected_canonical_url = f"http://oeo.net/openeo/{api.api_version}/jobs/{job_id}/results" + expected_self_url = f"http://oeo.net/openeo/{api.api_version}/jobs/{job_id}/results?partial=true" + expected_canonical_url = f"http://oeo.net/openeo/{api.api_version}/jobs/{job_id}/results?partial=true" assert job_result == DictSubSet( { "openeo:status": expected_openeo_status, "type": "Collection", "stac_version": "1.0.0", "id": job_id, - "title": "Unfinished batch job {job_id}", + "title": f"Unfinished batch job {job_id}", "description": f"Results for batch job {job_id}", "license": "proprietary", "links": [ + { + "rel": "self", + "href": expected_self_url, + "type": "application/json", + }, { "rel": "canonical", "href": expected_canonical_url, "type": "application/json", - } + }, ], } ) @@ -2443,6 +2449,7 @@ def test_get_job_results_signed_100_unfinished_and_partial_true( resp = api100.get(f"/jobs/{job_id}/results?partial=true", headers=self.AUTH_HEADER) resp.assert_status_code(200) + expected_self_url = f"http://oeo.net/openeo/1.0.0/jobs/{job_id}/results?partial=true" expected_canonical_url = f"http://oeo.net/openeo/1.0.0/jobs/{job_id}/results/TXIuVGVzdA==/05cb8b78f20c68a5aa9eb05249928d24?partial=true" assert resp.json == DictSubSet( { @@ -2450,10 +2457,15 @@ def test_get_job_results_signed_100_unfinished_and_partial_true( "type": "Collection", "stac_version": "1.0.0", "id": job_id, - "title": "Unfinished batch job {job_id}", + "title": f"Unfinished batch job {job_id}", "description": f"Results for batch job {job_id}", "license": "proprietary", "links": [ + { + "rel": "self", + "href": expected_self_url, + "type": "application/json", + }, { "rel": "canonical", "href": expected_canonical_url, @@ -2601,6 +2613,7 @@ def test_get_job_results_signed_110_unfinished_and_partial_true( ) resp.assert_status_code(200) + expected_self_url = f"http://oeo.net/openeo/1.1.0/jobs/{job_id}/results?partial=true" expected_canonical_url = f"http://oeo.net/openeo/1.1.0/jobs/{job_id}/results/TXIuVGVzdA==/05cb8b78f20c68a5aa9eb05249928d24?partial=true" assert resp.json == DictSubSet( { @@ -2608,10 +2621,15 @@ def test_get_job_results_signed_110_unfinished_and_partial_true( "type": "Collection", "stac_version": "1.0.0", "id": job_id, - "title": "Unfinished batch job {job_id}", + "title": f"Unfinished batch job {job_id}", "description": f"Results for batch job {job_id}", "license": "proprietary", "links": [ + { + "rel": "self", + "href": expected_self_url, + "type": "application/json", + }, { "rel": "canonical", "href": expected_canonical_url, @@ -2731,6 +2749,7 @@ def test_get_job_results_signed_with_expiration_100_unfinished_and_partial_true( resp = api100.get(f"/jobs/{job_id}/results?partial=true", headers=self.AUTH_HEADER) resp.assert_status_code(200) + expected_self_url = f"http://oeo.net/openeo/1.0.0/jobs/{job_id}/results?partial=true" expected_canonical_url = f"http://oeo.net/openeo/1.0.0/jobs/{job_id}/results/TXIuVGVzdA==/9fea29cd94195399cc4d902388a3c32c?expires=2234&partial=true" assert resp.json == DictSubSet( { @@ -2738,10 +2757,15 @@ def test_get_job_results_signed_with_expiration_100_unfinished_and_partial_true( "type": "Collection", "stac_version": "1.0.0", "id": job_id, - "title": "Unfinished batch job {job_id}", + "title": f"Unfinished batch job {job_id}", "description": f"Results for batch job {job_id}", "license": "proprietary", "links": [ + { + "rel": "self", + "href": expected_self_url, + "type": "application/json", + }, { "rel": "canonical", "href": expected_canonical_url, @@ -2886,6 +2910,7 @@ def test_get_job_results_signed_with_expiration_110_unfinished_and_partial_true( resp = api110.get(f"/jobs/{job_id}/results?partial=true", headers=self.AUTH_HEADER) resp.assert_status_code(200) + expected_self_url = f"http://oeo.net/openeo/1.1.0/jobs/{job_id}/results?partial=true" expected_canonical_url = f"http://oeo.net/openeo/1.1.0/jobs/{job_id}/results/TXIuVGVzdA==/9fea29cd94195399cc4d902388a3c32c?expires=2234&partial=true" assert resp.json == DictSubSet( { @@ -2893,10 +2918,15 @@ def test_get_job_results_signed_with_expiration_110_unfinished_and_partial_true( "type": "Collection", "stac_version": "1.0.0", "id": job_id, - "title": "Unfinished batch job {job_id}", + "title": f"Unfinished batch job {job_id}", "description": f"Results for batch job {job_id}", "license": "proprietary", "links": [ + { + "rel": "self", + "href": expected_self_url, + "type": "application/json", + }, { "rel": "canonical", "href": expected_canonical_url,