feat(core): STAC formatted serialization

eodag/api/collection.py

@@ -20,7 +20,7 @@
 import logging
 import re
 from collections import UserDict, UserList
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Optional, cast
 
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr
 from pydantic import ValidationError as PydanticValidationError
@@ -29,6 +29,9 @@
 from stac_pydantic.collection import Extent, Provider, SpatialExtent, TimeInterval
 from stac_pydantic.links import Links
 
+from eodag.types.queryables import CommonStacMetadata
+from eodag.types.stac_metadata import create_stac_metadata_model
+from eodag.utils import STAC_VERSION
 from eodag.utils.env import is_env_var_true
 from eodag.utils.exceptions import ValidationError
 from eodag.utils.repr import dict_to_html_table
@@ -281,6 +284,62 @@ def list_queryables(self, **kwargs: Any) -> QueryablesDict:
 
         return dag.list_queryables(collection=self.id, **kwargs)
 
+    def serialize(self) -> dict[str, Any]:
+        """Serialize the Collection instance to a STAC dictionary.
+
+        :returns: A STAC dictionary representation of the Collection instance.
+        """
+        stac_dict: dict[str, Any] = {
+            "stac_version": STAC_VERSION,
+            "type": "Collection",
+        }
+
+        stac_dict |= self.model_dump(mode="json", exclude_none=True, exclude={"alias"})
+
+        stac_dict.setdefault("links", [])
+        stac_dict.setdefault("providers", [])
+
+        not_in_summaries = [
+            "stac_version",
+            "type",
+            "id",
+            "title",
+            "description",
+            "extent",
+            "keywords",
+            "license",
+            "links",
+            "providers",
+        ]
+        summaries = dict()
+        for k, v in stac_dict.items():
+            if k not in not_in_summaries:
+                if isinstance(v, list):
+                    summaries[k] = v
+                elif isinstance(v, str):
+                    summaries[k] = v.split(",")
+                else:
+                    summaries[k] = [v]
+        stac_dict["summaries"] = summaries
+
+        # Remove empty items and items moved to summaries
+        keys_to_remove = [
+            k
+            for k in stac_dict.keys()
+            if k not in not_in_summaries and k != "summaries"
+        ]
+        for k in keys_to_remove:
+            del stac_dict[k]
+
+        # add extensions
+        summaries_model = cast(CommonStacMetadata, create_stac_metadata_model())
+        summaries_validated = summaries_model.model_construct(
+            _fields_set=None, **summaries
+        )
+        stac_dict["stac_extensions"] = summaries_validated.get_conformance_classes()
+
+        return stac_dict
+
 
 class CollectionsDict(UserDict[str, Collection]):
     """A UserDict object which values are :class:`~eodag.api.collection.Collection` objects, keyed by provider ``id``.

eodag/api/core.py

@@ -30,6 +30,7 @@
 from importlib.metadata import version
 from importlib.resources import files as res_files
 from operator import attrgetter, itemgetter
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Iterator, Optional, Union, cast
 
 import geojson
@@ -1783,14 +1784,15 @@ def _do_search(
 
             # remove None values and convert param names to their pydantic alias if any
             search_params = {}
+            queryables_fields = Queryables.from_stac_models().model_fields
             ecmwf_queryables = [
                 f"{ECMWF_PREFIX[:-1]}_{k}" for k in ECMWF_ALLOWED_KEYWORDS
             ]
             for param, value in kwargs.items():
                 if value is None:
                     continue
-                if param in Queryables.model_fields:
-                    param_alias = Queryables.model_fields[param].alias or param
+                if param in queryables_fields:
+                    param_alias = queryables_fields[param].alias or param
                     search_params[param_alias] = value
                 elif param in ecmwf_queryables:
                     # alias equivalent for ECMWF queryables
@@ -1985,8 +1987,42 @@ def serialize(
         :param filename: (optional) The name of the file to generate
         :returns: The name of the created file
         """
+        search_result_dict = search_result.as_geojson_object()
+        # add self link
+        search_result_dict.setdefault("links", [])
+        search_result_dict["links"].append(
+            {
+                "rel": "self",
+                "href": f"{filename}",
+                "type": "application/json",
+            },
+        )
+        # write search results
         with open(filename, "w") as fh:
-            geojson.dump(search_result.as_geojson_object(), fh)
+            geojson.dump(search_result_dict, fh)
+            logger.debug("Search results saved to %s", filename)
+        # write collection(s)
+        if search_result._dag is None:
+            return filename
+        collections = set(p.collection for p in search_result)
+        for collection in collections:
+            collection_obj = search_result._dag.collections_config.get(
+                collection, Collection(id=collection)
+            )
+            collection_dict = collection_obj.serialize()
+            # add links
+            collection_dict.setdefault("links", [])
+            collection_dict["links"].append(
+                {
+                    "rel": "self",
+                    "href": f"{collection}.json",
+                    "type": "application/json",
+                },
+            )
+            with open(Path(filename).parent / f"{collection}.json", "w") as fh:
+                geojson.dump(collection_dict, fh)
+                logger.debug("Collection '%s' saved to %s", collection, fh.name)
+
         return filename
 
     @staticmethod
@@ -2190,7 +2226,8 @@ def list_queryables(
 
             # use queryables aliases
             kwargs_alias = {**kwargs}
-            for search_param, field_info in Queryables.model_fields.items():
+            queryables_fields = Queryables.from_stac_models().model_fields
+            for search_param, field_info in queryables_fields.items():
                 if search_param in kwargs and field_info.alias:
                     kwargs_alias[field_info.alias] = kwargs_alias.pop(search_param)
 

eodag/api/product/_product.py

@@ -23,14 +23,18 @@
 import re
 import tempfile
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union, cast
 
+import orjson
 import requests
 from requests import RequestException
 from requests.auth import AuthBase
 from shapely import geometry
 from shapely.errors import ShapelyError
 
+from eodag.types.queryables import CommonStacMetadata
+from eodag.types.stac_metadata import create_stac_metadata_model
+
 try:
     # import from eodag-cube if installed
     from eodag_cube.api.product import (  # pyright: ignore[reportMissingImports]
@@ -52,6 +56,7 @@
     DEFAULT_DOWNLOAD_WAIT,
     DEFAULT_SHAPELY_GEOMETRY,
     DEFAULT_STREAM_REQUESTS_TIMEOUT,
+    STAC_VERSION,
     USER_AGENT,
     ProgressCallback,
     format_string,
@@ -146,6 +151,13 @@ def __init__(
             and not key.startswith("_")
             and value is not None
         }
+        self.properties.setdefault(
+            "datetime",
+            self.properties.get("start_datetime")
+            or self.properties.get("end_datetime"),
+        )
+
+        # sort properties to have common stac properties first
         common_stac_properties = {
             key: self.properties[key]
             for key in sorted(self.properties)
@@ -205,25 +217,45 @@ def as_dict(self) -> dict[str, Any]:
         """
         search_intersection = None
         if self.search_intersection is not None:
-            search_intersection = geometry.mapping(self.search_intersection)
+            search_intersection = orjson.loads(
+                orjson.dumps(self.search_intersection.__geo_interface__)
+            )
+
+        stac_properties = {
+            **{
+                key: value
+                for key, value in self.properties.items()
+                if key not in ("geometry", "id")
+            },
+            "eodag:provider": self.provider,
+            "eodag:search_intersection": search_intersection,
+        }
+        stac_providers = self.properties.get("providers", [])
+        if not any("host" in p.get("roles", []) for p in stac_providers):
+            stac_providers.append({"name": self.provider, "roles": ["host"]})
+            stac_properties["providers"] = stac_providers
+
+        props_model = cast(CommonStacMetadata, create_stac_metadata_model())
+        props_validated = props_model.model_validate(stac_properties)
 
         geojson_repr: dict[str, Any] = {
             "type": "Feature",
-            "geometry": geometry.mapping(self.geometry),
+            "geometry": orjson.loads(orjson.dumps(self.geometry.__geo_interface__)),
+            "bbox": list(self.geometry.bounds),
             "id": self.properties["id"],
             "assets": self.assets.as_dict(),
-            "properties": {
-                "eodag:collection": self.collection,
-                "eodag:provider": self.provider,
-                "eodag:search_intersection": search_intersection,
-                **{
-                    key: value
-                    for key, value in self.properties.items()
-                    if key not in ("geometry", "id")
+            "properties": stac_properties,
+            "links": [
+                {
+                    "rel": "collection",
+                    "href": f"{self.collection}.json",
+                    "type": "application/json",
                 },
-            },
+            ],
+            "stac_extensions": props_validated.get_conformance_classes(),
+            "stac_version": STAC_VERSION,
+            "collection": self.collection,
         }
-
         return geojson_repr
 
     @classmethod
@@ -237,11 +269,11 @@ def from_geojson(cls, feature: dict[str, Any]) -> EOProduct:
         :raises: :class:`~eodag.utils.exceptions.ValidationError`
         """
         try:
+            collection = feature.get("collection")
             properties = feature["properties"]
             properties["geometry"] = feature["geometry"]
             properties["id"] = feature["id"]
             provider = properties.pop("eodag:provider")
-            collection = properties.pop("eodag:collection")
             search_intersection = properties.pop("eodag:search_intersection")
         except KeyError as e:
             raise ValidationError(

eodag/api/product/metadata_mapping.py

@@ -1731,16 +1731,19 @@ def get_queryable_from_provider(
     mapping_values = [
         v[0] if isinstance(v, list) else "" for v in metadata_mapping.values()
     ]
+    StacQueryables = Queryables.from_stac_models()
     if provider_queryable in mapping_values:
         ind = mapping_values.index(provider_queryable)
-        return Queryables.get_queryable_from_alias(list(metadata_mapping.keys())[ind])
+        return StacQueryables.get_queryable_from_alias(
+            list(metadata_mapping.keys())[ind]
+        )
     for param, param_conf in metadata_mapping.items():
         if (
             isinstance(param_conf, list)
             and param_conf[0]
             and re.search(pattern, param_conf[0])
         ):
-            return Queryables.get_queryable_from_alias(param)
+            return StacQueryables.get_queryable_from_alias(param)
     return None
 
 

eodag/api/search_result.py

@@ -32,7 +32,7 @@
 from eodag.plugins.crunch.filter_latest_tpl_name import FilterLatestByName
 from eodag.plugins.crunch.filter_overlap import FilterOverlap
 from eodag.plugins.crunch.filter_property import FilterProperty
-from eodag.utils import GENERIC_STAC_PROVIDER, STAC_SEARCH_PLUGINS
+from eodag.utils import GENERIC_STAC_PROVIDER, STAC_SEARCH_PLUGINS, STAC_VERSION
 from eodag.utils.exceptions import MisconfiguredError
 
 if TYPE_CHECKING:
@@ -206,6 +206,9 @@ def as_geojson_object(self) -> dict[str, Any]:
                 "eodag:search_params": geojson_search_params or None,
                 "eodag:raise_errors": self.raise_errors,
             },
+            "links": [],
+            "stac_extensions": [],
+            "stac_version": STAC_VERSION,
         }
 
     def as_shapely_geometry_object(self) -> GeometryCollection:

eodag/plugins/search/base.py

@@ -35,6 +35,7 @@
 from eodag.types import model_fields_to_annotated
 from eodag.types.queryables import Queryables, QueryablesDict
 from eodag.types.search_args import SortByList
+from eodag.types.stac_metadata import CommonStacMetadata, create_stac_metadata_model
 from eodag.utils import (
     GENERIC_COLLECTION,
     copy_deepcopy,
@@ -358,7 +359,7 @@ def _get_collection_queryables(
             queryables = self.discover_queryables(**{**default_values, **filters}) or {}
         except NotImplementedError as e:
             if str(e):
-                logger.debug(str(e))
+                logger.debug("%s, configured metadata-mapping used", str(e))
             queryables = self.queryables_from_metadata_mapping(collection, alias)
 
         return QueryablesDict(**queryables)
@@ -408,9 +409,10 @@ def list_queryables(
                 col_queryables = self._get_collection_queryables(col, None, filters)
                 all_queryables.update(col_queryables)
             # reset defaults because they may vary between collections
+            queryables_fields = Queryables.from_stac_models().model_fields
             for k, v in all_queryables.items():
                 v.__metadata__[0].default = getattr(
-                    Queryables.model_fields.get(k, Field(None)), "default", None
+                    queryables_fields.get(k, Field(None)), "default", None
                 )
             return QueryablesDict(
                 additional_properties=auto_discovery,
@@ -468,8 +470,11 @@ def queryables_from_metadata_mapping(
             ):
                 del metadata_mapping[param]
 
+        queryables_model = create_stac_metadata_model(
+            base_models=[Queryables, CommonStacMetadata]
+        )
         eodag_queryables = copy_deepcopy(
-            model_fields_to_annotated(Queryables.model_fields)
+            model_fields_to_annotated(queryables_model.model_fields)
         )
         queryables["collection"] = eodag_queryables.pop("collection")
         # add default value for collection

eodag/plugins/search/build_search_result.py

@@ -824,9 +824,7 @@ def discover_queryables(
                 )
                 not in set(list(available_values.keys()) + [f["name"] for f in form])
             ):
-                raise ValidationError(
-                    f"'{keyword}' is not a queryable parameter", {keyword}
-                )
+                raise ValidationError("'%s' is not a queryable parameter" % keyword)
 
         # generate queryables
         if form: