COPDS-2954: stac validation at ingestion level (#170)

* feat: stac validation at ingestion level

* chore: resource stac-validation

* chore: integrated new get_extent error handling

Author: alfredotoselli

cads_catalogue/entry_points.py

@@ -285,6 +285,7 @@ def update_catalogue(
     exclude_licences: bool = False,
     exclude_messages: bool = False,
     exclude_contents: bool = False,
+    perform_stac_validation: bool = True,
 ) -> None:
     """Update the database with the catalogue data.
 
@@ -306,6 +307,7 @@ def update_catalogue(
     :param exclude_licences: if True, do not consider input licences (default False)
     :param exclude_messages: if True, do not consider input messages (default False)
     :param exclude_contents: if True, do not consider input contents (default False)
+    :param perform_stac_validation: if False, skip STAC compliance validation
     """
     cads_common.logging.structlog_configure()
     cads_common.logging.logging_configure()
@@ -366,6 +368,7 @@ def update_catalogue(
                 include=include,
                 exclude=exclude,
                 override_md=new_catalogue_update_md["override_md"],
+                perform_stac_validation=perform_stac_validation,
             )
         if "messages" in to_process:
             logger.info("db updating of messages")

cads_catalogue/manager.py

@@ -37,6 +37,7 @@
     layout_manager,
     object_storage,
     utils,
+    validations,
 )
 
 logger = structlog.get_logger(__name__)
@@ -818,6 +819,7 @@ def update_catalogue_resources_single_folder(
     include: List[str] = [],
     exclude: List[str] = [],
     override_md: dict[str, Any] = {},
+    perform_stac_validation: bool = True,
 ) -> List[str]:
     """
     Load metadata of resources from files of a single input folder and sync each resource in the db.
@@ -832,6 +834,7 @@ def update_catalogue_resources_single_folder(
     include: list of include patterns for the resource uids
     exclude: list of exclude patterns for the resource uids
     override_md: dictionary of override metadata for resources
+    perform_stac_validation: if False, skip STAC compliance validation
 
     Returns
     -------
@@ -887,6 +890,10 @@ def update_catalogue_resources_single_folder(
                     session, resource_folder_path, resource, storage_settings
                 )
                 resource["adaptor_properties_hash"] = compute_config_hash(resource)
+
+                if perform_stac_validation:
+                    validations.validate_resource_dict_stac_compliance(resource)
+
                 resource_sync(session, resource, storage_settings)
             logger.info("resource '%s' db sync successful" % resource_uid)
         except Exception:  # noqa
@@ -905,6 +912,7 @@ def update_catalogue_resources(
     include: List[str] = [],
     exclude: List[str] = [],
     override_md: dict[str, Any] = {},
+    perform_stac_validation: bool = False,
 ) -> List[str]:
     """
     Load metadata of resources from files and sync each resource in the db.
@@ -919,6 +927,7 @@ def update_catalogue_resources(
     include: list of include patterns for the resource uids
     exclude: list of exclude patterns for the resource uids
     override_md: dictionary of override metadata for resources
+    perform_stac_validation: if False, skip STAC compliance validation
 
     Returns
     -------
@@ -935,6 +944,7 @@ def update_catalogue_resources(
             include,
             exclude,
             override_md,
+            perform_stac_validation,
         )
         involved_resource_uids += new_involved
     return involved_resource_uids

cads_catalogue/stac_helpers.py

@@ -0,0 +1,94 @@
+"""STAC helpers."""
+
+# Copyright 2025, European Union.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+
+import pydantic
+import stac_pydantic
+import structlog
+
+from . import database
+
+logger = structlog.getLogger(__name__)
+
+
+WRONG_BBOX_LOGGED_IDS: set[str] = set()
+
+
+def get_extent(
+    record: database.Resource,
+) -> dict:
+    """Get extent from model."""
+    spatial = record.geo_extent or {}
+
+    west = float(spatial.get("bboxW", -180))
+    south = float(spatial.get("bboxS", -90))
+    east = float(spatial.get("bboxE", 180))
+    north = float(spatial.get("bboxN", 90))
+
+    try:
+        spatial_extent = stac_pydantic.collection.SpatialExtent(
+            bbox=[(west, south, east, north)],
+        )
+    except pydantic.ValidationError as e:
+        # 0-360 longitude values are considered valid
+        if (
+            0 <= west <= 360 or 0 <= east <= 360
+        ) and "Bounding box must be within (-180, -90, 180, 90)" in str(e):
+            spatial_extent = stac_pydantic.collection.SpatialExtent.model_construct(
+                bbox=[(west, south, east, north)]
+            )
+        else:
+            if record.resource_uid not in WRONG_BBOX_LOGGED_IDS:
+                # this log is important, but we don't want to flood the logs
+                WRONG_BBOX_LOGGED_IDS.add(str(record.resource_uid))
+                logger.warning(
+                    "Bbox stac_pydantic validation failed, fallback to whole world",
+                    error=e,
+                    id=record.resource_uid,
+                )
+            spatial_extent = stac_pydantic.collection.SpatialExtent(
+                bbox=[(-180, -90, 180, 90)]
+            )
+
+    begin_date_value = getattr(record, "begin_date", None)
+    end_date_value = getattr(record, "end_date", None)
+
+    return stac_pydantic.collection.Extent(
+        spatial=spatial_extent,
+        temporal=stac_pydantic.collection.TimeInterval(
+            interval=[
+                [
+                    (
+                        # We have datetime.date on DB, while STAC requires datetime with timezone
+                        datetime.datetime.combine(
+                            begin_date_value, datetime.datetime.min.time()
+                        ).replace(tzinfo=datetime.timezone.utc)
+                        if begin_date_value
+                        else None
+                    ),
+                    (
+                        # We have datetime.date on DB, while STAC requires datetime with timezone
+                        datetime.datetime.combine(
+                            end_date_value, datetime.datetime.min.time()
+                        ).replace(tzinfo=datetime.timezone.utc)
+                        if end_date_value
+                        else None
+                    ),
+                ]
+            ],
+        ),
+    ).model_dump()

cads_catalogue/validations.py

@@ -23,8 +23,9 @@
 from typing import Any
 
 import cads_common.logging
+import pydantic
 
-from cads_catalogue import layout_manager, utils
+from cads_catalogue import database, layout_manager, stac_helpers, utils
 
 logger = logging.getLogger(__name__)
 
@@ -470,3 +471,40 @@ def validate_datasets(datasets_folder: str, loglevel: str = "info") -> None:
             continue
         validate_dataset(dataset_folder, loglevel=None)
     print("----end of validations----")
+
+
+def validate_resource_dict_stac_compliance(
+    resource_dict: dict[str, Any],
+) -> None:
+    """Validate that resource metadata dictionary will be STAC-compliant."""
+    resource_uid = resource_dict.get("resource_uid", "unknown")
+
+    begin_date = resource_dict.get("begin_date")
+    if isinstance(begin_date, str):
+        begin_date = datetime.date.fromisoformat(begin_date) if begin_date else None
+
+    end_date = resource_dict.get("end_date")
+    if isinstance(end_date, str):
+        end_date = datetime.date.fromisoformat(end_date) if end_date else None
+
+    resource_model = database.Resource(
+        resource_uid=resource_uid,
+        geo_extent=resource_dict.get("geo_extent"),
+        begin_date=begin_date,
+        end_date=end_date,
+    )
+
+    try:
+        _ = stac_helpers.get_extent(resource_model)
+    except pydantic.ValidationError as e:
+        logger.error(
+            f"STAC validation failed for resource dictionary '{resource_uid}': {e.errors()}"
+        )
+        raise
+    except Exception as e:
+        logger.error(
+            f"Unexpected error during STAC validation for resource '{resource_uid}': {e}"
+        )
+        raise ValueError(f"STAC validation failed: {e}") from e
+
+    logger.info(f"STAC validation passed for resource dictionary '{resource_uid}'")

environment.yml

@@ -11,10 +11,13 @@ dependencies:
 - boto3
 - gitpython
 - jsonschema
+- pydantic
 - psycopg2
 - python-frontmatter
 - rfc3339-validator
 - sqlalchemy>=2.0.9
 - sqlalchemy-utils
 - structlog
 - typer
+- pip:
+  - stac_pydantic

pyproject.toml

@@ -21,6 +21,8 @@ dependencies = [
   "cads-e2e-tests@git+https://github.com/ecmwf-projects/cads-e2e-tests.git",
   "gitpython",
   "jsonschema",
+  "pydantic",
+  "stac-pydantic",
   "python-frontmatter",
   "rfc3339-validator",
   "sqlalchemy>=2.0.9",