feat: implement polling mechanism (#5)

implements a polling mechanism for products that have to be ordered:

- When a search is done for an orderable product, no `assets` are
returned but in the `links` section, a `retrieve` link with a POST body
is given. With the link the product can be ordered.
- The status of the order can be check using the `self` link.
- When the order is ready, the `downloadLink` is available.

---------

Co-authored-by: LAMBARE Aubin <aubin.lambare@cs-soprasteria.com>
Co-authored-by: LAMBARE Aubin <aubin.lambare@csgroup.eu>
Co-authored-by: jlahovnik <julia.lahovnik@csgroup.eu>

Author: 3 people

README.md

@@ -5,7 +5,6 @@
   <img src="https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png" alt="FastAPI" height=100 />
 </p>
 
-
 [EODAG](https://github.com/CS-SI/eodag) backend for [stac-fastapi](https://github.com/stac-utils/stac-fastapi), the [FastAPI](https://fastapi.tiangolo.com/) implementation of the [STAC API spec](https://github.com/radiantearth/stac-api-spec)
 
 stac-fastapi-eodag combines the capabilities of EODAG and STAC FastAPI to provide a powerful, unified API for accessing Earth observation data from various providers.

stac_fastapi/eodag/constants.py

@@ -19,8 +19,12 @@
 
 ITEM_PROPERTIES_EXCLUDE = {
     "_id": True,
+    "productType": True,
     "downloadLink": True,
     "orderLink": True,
+    "orderStatus": True,
+    "orderStatusLink": True,
+    "searchLink": True,
     "missionStartDate": True,
     "missionEndDate": True,
     "keywords": True,

stac_fastapi/eodag/core.py

@@ -44,11 +44,12 @@
 
 from eodag import SearchResult
 from eodag.api.core import DEFAULT_ITEMS_PER_PAGE
+from eodag.plugins.search.build_search_result import ECMWFSearch
 from eodag.utils import deepcopy, get_geometry_from_various
-from eodag.utils.exceptions import NoMatchingProductType
+from eodag.utils.exceptions import NoMatchingProductType as EodagNoMatchingProductType
 from stac_fastapi.eodag.config import get_settings
 from stac_fastapi.eodag.cql_evaluate import EodagEvaluator
-from stac_fastapi.eodag.errors import ResponseSearchError
+from stac_fastapi.eodag.errors import NoMatchingProductType, ResponseSearchError
 from stac_fastapi.eodag.landing_page import CustomCoreClient
 from stac_fastapi.eodag.models.links import (
     CollectionLinks,
@@ -132,10 +133,25 @@ def _get_collection(self, product_type: dict[str, Any], request: Request) -> Col
 
         collection["id"] = product_type["ID"]
 
+        # keep only federation backends which allow order mechanism
+        # to create "retrieve" collection links from them
+        def has_ecmwf_search_plugin(federation_backends, request):
+            for fb in federation_backends:
+                search_plugins = request.app.state.dag._plugins_manager.get_search_plugins(provider=fb)
+                if any(isinstance(plugin, ECMWFSearch) for plugin in search_plugins):
+                    return True
+            return False
+
         extension_names = [type(ext).__name__ for ext in self.extensions]
-        collection["links"] = CollectionLinks(collection_id=collection["id"], request=request).get_links(
-            extensions=extension_names, extra_links=product_type.get("links", []) + collection.get("links", [])
-        )
+        if self.extension_is_enabled("CollectionOrderExtension") and not has_ecmwf_search_plugin(
+            federation_backends, request
+        ):
+            extension_names.remove("CollectionOrderExtension")
+
+        collection["links"] = CollectionLinks(
+            collection_id=collection["id"],
+            request=request,
+        ).get_links(extensions=extension_names, extra_links=product_type.get("links", []) + collection.get("links", []))
 
         collection["providers"] = merge_providers(
             collection.get("providers", []) + [get_provider_dict(request, fb) for fb in federation_backends]
@@ -144,40 +160,42 @@ def _get_collection(self, product_type: dict[str, Any], request: Request) -> Col
         return collection
 
     def _search_base(self, search_request: BaseSearchPostRequest, request: Request) -> ItemCollection:
+        eodag_args = prepare_search_base_args(search_request=search_request, model=self.stac_metadata_model)
+
+        request.state.eodag_args = eodag_args
+
         # check if the collection exists
-        if search_request.collections:
+        if product_type := eodag_args.get("productType"):
             all_pt = request.app.state.dag.list_product_types(fetch_providers=False)
             # only check the first collection (EODAG search only support a single collection)
-            existing_pt = [pt for pt in all_pt if pt["ID"] == search_request.collections[0]]
+            existing_pt = [pt for pt in all_pt if pt["ID"] == product_type]
             if not existing_pt:
-                raise NoMatchingProductType(f"Collection {search_request.collections[0]} does not exist.")
+                raise NoMatchingProductType(f"Collection {product_type} does not exist.")
         else:
             raise HTTPException(status_code=400, detail="A collection is required")
 
         # get products by ids
-        if search_request.ids:
+        if ids := eodag_args.pop("ids", []):
             search_result = SearchResult([])
-            ids = search_request.ids
             for item_id in ids:
-                search_request.ids = [item_id]
-                base_args = prepare_search_base_args(search_request=search_request, model=self.stac_metadata_model)
-                search_result.extend(request.app.state.dag.search(**base_args))
+                eodag_args["id"] = item_id
+                search_result.extend(request.app.state.dag.search(**eodag_args))
             search_result.number_matched = len(search_result)
         else:
             # search without ids
-            base_args = prepare_search_base_args(search_request=search_request, model=self.stac_metadata_model)
-            search_result = request.app.state.dag.search(**base_args)
+            search_result = request.app.state.dag.search(**eodag_args)
 
         if search_result.errors and not len(search_result):
             raise ResponseSearchError(search_result.errors, self.stac_metadata_model)
 
         request_json = loop.run_until_complete(request.json()) if request.method == "POST" else None
 
         features: list[Item] = []
+        extension_names = [type(ext).__name__ for ext in self.extensions]
 
         for product in search_result:
             feature = create_stac_item(
-                product, self.stac_metadata_model, self.extension_is_enabled, request, request_json
+                product, self.stac_metadata_model, self.extension_is_enabled, request, extension_names, request_json
             )
             features.append(feature)
 
@@ -198,7 +216,6 @@ def _search_base(self, search_request: BaseSearchPostRequest, request: Request)
             ):
                 next_page = search_request.page + 1
 
-        extension_names = [type(ext).__name__ for ext in self.extensions]
         collection["links"] = PagingLinks(
             request=request,
             next=next_page,
@@ -246,7 +263,7 @@ async def all_collections(
                 guessed_product_types = request.app.state.dag.guess_product_type(
                     free_text=q, missionStartDate=start, missionEndDate=end
                 )
-            except NoMatchingProductType:
+            except EodagNoMatchingProductType:
                 product_types = []
             else:
                 product_types = [pt for pt in all_pt if pt["ID"] in guessed_product_types]
@@ -416,11 +433,11 @@ def get_search(
 
         if filter_expr:
             if filter_lang == "cql2-text":
-                ast = parse_cql2_text(filter_expr)
-                base_args["filter_expr"] = str2json("filter_expr", to_cql2(ast))  # type: ignore
-                base_args["filter-lang"] = "cql2-json"
-            elif filter_lang == "cql-json":
-                base_args["filter_expr"] = str2json(filter_expr)
+                filter_expr = to_cql2(parse_cql2_text(filter_expr))
+                filter_lang = "cql2-json"
+
+            base_args["filter"] = str2json("filter_expr", filter_expr)
+            base_args["filter_lang"] = "cql2-json"
 
         # Remove None values from dict
         clean = {}
@@ -433,7 +450,7 @@ def get_search(
         except ValidationError as err:
             raise HTTPException(status_code=400, detail=f"Invalid parameters provided {err}") from err
 
-        return self.post_search(search_request, request)
+        return self._search_base(search_request, request)
 
     async def get_item(self, item_id: str, collection_id: str, request: Request, **kwargs: Any) -> Item:
         """
@@ -555,9 +572,8 @@ def prepare_search_base_args(search_request: BaseSearchPostRequest, model: type[
     if search_request.collections:
         base_args["productType"] = search_request.collections[0]
 
-    # handle only one id from here (pre-filtered in _search_base)
     if search_request.ids:
-        base_args["id"] = search_request.ids[0]
+        base_args["ids"] = search_request.ids
 
     # merge all eodag search arguments
     base_args = base_args | sort_by | eodag_filter | eodag_query

stac_fastapi/eodag/eodag_types/queryables.py

@@ -28,6 +28,7 @@ class QueryablesGetParams(BaseModel):
     """Store GET Queryables query params"""
 
     collection: Optional[str] = Field(default=None, serialization_alias="productType")
+    provider: Optional[str] = Field(default=None)
     start_datetime: Union[list[str], str, None] = Field(default=None)
     end_datetime: Union[list[str], str, None] = Field(default=None)
     datetime: Union[list[str], str, None] = Field(default=None)

stac_fastapi/eodag/extensions/collection_order.py

@@ -20,23 +20,22 @@
 import logging
 from typing import (
     Annotated,
+    Optional,
     cast,
 )
 
 import attr
 from eodag.api.core import EODataAccessGateway
 from eodag.api.product._product import EOProduct
 from eodag.api.product.metadata_mapping import OFFLINE_STATUS
-from fastapi import APIRouter, FastAPI, Path, Query, Request
-from stac_fastapi.api.routes import create_async_endpoint
-from stac_fastapi.types.errors import NotFoundError
+from fastapi import APIRouter, Depends, FastAPI, Path, Request
+from pydantic import BaseModel, ConfigDict, Field
+from stac_fastapi.api.errors import NotFoundError
+from stac_fastapi.api.routes import _wrap_response, sync_to_async
 from stac_fastapi.types.extension import ApiExtension
 from stac_fastapi.types.search import APIRequest
 from stac_fastapi.types.stac import Item
 
-from stac_fastapi.eodag.errors import (
-    MisconfiguredError,
-)
 from stac_fastapi.eodag.models.stac_metadata import (
     CommonStacMetadata,
     create_stac_item,
@@ -45,6 +44,20 @@
 logger = logging.getLogger(__name__)
 
 
+class CollectionOrderBody(BaseModel):
+    """Collection order request body."""
+
+    federation_backends: Optional[list[str]] = Field(
+        default=None,
+        description=(
+            "Federation backends filter. Default is None which means no filter is applied. Only one value is supported"
+        ),
+        alias="federation:backends",
+    )
+
+    model_config = ConfigDict(extra="allow", json_schema_extra={"examples": [{"date": "string", "variable": "string"}]})
+
+
 @attr.s
 class BaseCollectionOrderClient:
     """Defines a pattern for implementing the collection order extension."""
@@ -58,16 +71,18 @@ def extension_is_enabled(self, extension: str) -> bool:
 
     def order_collection(
         self,
-        federation_backend: str,
         collection_id: str,
-        dc_qs: str,
         request: Request,
+        request_body: CollectionOrderBody,
     ) -> Item:
         """Order a product with its collection id and a fake id"""
 
         dag = cast(EODataAccessGateway, request.app.state.dag)
 
-        search_results = dag.search(id="fake_id", productType=collection_id, provider=federation_backend, _dc_qs=dc_qs)
+        federation_backend = request_body.federation_backends[0] if request_body.federation_backends else None
+
+        request_params = request_body.model_dump(exclude={"federation_backends": True})
+        search_results = dag.search(productType=collection_id, provider=federation_backend, **request_params)
         if len(search_results) > 0:
             product = cast(EOProduct, search_results[0])
 
@@ -76,62 +91,51 @@ def order_collection(
                 f"Could not find any item in {collection_id} collection for backend {federation_backend}.",
             )
 
-        if not getattr(product.downloader, "order_download", None):
-            raise MisconfiguredError("Product downloader must have a the order method")
-
         auth = product.downloader_auth.authenticate() if product.downloader_auth else None
 
         if product.properties.get("orderLink") is None or product.properties.get("storageStatus") != OFFLINE_STATUS:
             raise NotFoundError(
                 "Product is not orderable. Please download it directly.",
             )
 
-        if product.properties.get("orderStatus"):
-            raise NotFoundError(
-                "Product has been ordered previously. Please request the polling endpoint before download it."
-            )
-
         raise_error = False
         if product.downloader is None:
             logger.error("No downloader available for %s", product)
             raise_error = True
-
-        elif not hasattr(product.downloader, "order_download"):
-            logger.error("No order_download method available for %s of %s", product.downloader, product)
+        elif not hasattr(product.downloader, "order"):
+            logger.error("No order() method available for %s of %s", product.downloader, product)
             raise_error = True
         else:
             logger.debug("Order product")
-            _ = product.downloader.order_download(product=product, auth=auth)
+            product.downloader.order(product=product, auth=auth, timeout=-1)
 
         if raise_error or product.properties.get("orderId") is None:
             raise NotFoundError(
                 "Download order failed. It can be due to a lack of product found, so you "
-                f"may change 'dc_qs' argument. The one used for this order was: {dc_qs}"
+                "may change the body of the request."
             )
+        extension_names = [type(ext).__name__ for ext in self.extensions]
 
-        return create_stac_item(product, self.stac_metadata_model, self.extension_is_enabled, request)
+        return create_stac_item(product, self.stac_metadata_model, self.extension_is_enabled, request, extension_names)
 
 
 @attr.s
 class CollectionOrderUri(APIRequest):
     """Order collection."""
 
-    federation_backend: Annotated[str, Path(description="Federation backend name")] = attr.ib()
     collection_id: Annotated[str, Path(description="Collection ID")] = attr.ib()
-    dc_qs: Annotated[str, Query(description="Datacube query string")] = attr.ib()
 
 
 @attr.s
 class CollectionOrderExtension(ApiExtension):
     """Collection Order extension.
 
-    The order-collection extension allow to order a collection directly through the EODAG STAC
-    server.
+    The order-collection extension allow to order a collection directly through the EODAG STAC server.
 
     Usage:
     ------
 
-        ``POST /collections/{collection_id}/{federation_backend}/orders``
+        ``POST /collections/{collection_id}/order``
     """
 
     client: BaseCollectionOrderClient = attr.ib(factory=BaseCollectionOrderClient)
@@ -144,10 +148,20 @@ def register(self, app: FastAPI) -> None:
         :param app: Target FastAPI application.
         :returns: None
         """
+        func = sync_to_async(self.client.order_collection)
+
+        async def _retrieve_endpoint(
+            request: Request,
+            request_data: CollectionOrderBody,
+            request_path: CollectionOrderUri = Depends(),  # noqa: B008
+        ):
+            """Retrieve endpoint."""
+            return _wrap_response(await func(request=request, request_body=request_data, **request_path.kwargs()))
+
         self.router.prefix = app.state.router_prefix
         self.router.add_api_route(
             name="Order collection",
-            path="/collections/{collection_id}/{federation_backend}/orders",
+            path="/collections/{collection_id}/order",
             methods=["POST"],
             responses={
                 200: {
@@ -156,6 +170,6 @@ def register(self, app: FastAPI) -> None:
                     },
                 }
             },
-            endpoint=create_async_endpoint(self.client.order_collection, CollectionOrderUri),
+            endpoint=_retrieve_endpoint,
         )
         app.include_router(self.router, tags=["Collection order"])