RSPY578: fix cadip limit (#816)

* First implementation

* Fix UT

* Revert search tests

* Implemented parralellisation?

* Revert this config

* fix fmt

* Sort after regroup

* Fixed limit value

* removed unneccesary sort

* Review refactor

* Sonar fix

* Removed indexing

* Moved some func

* clean: code

* fix: cadip Item formating

* fix: cadip Item formating

---------

Co-authored-by: Julien Gaucher <[email protected]>

Author: Padeanu

services/adgs/rs_server_adgs/api/adgs_search.py

@@ -100,7 +100,7 @@ def process_search(
         limit: int,
         page: int,
     ) -> stac_pydantic.ItemCollection:
-        """Do the search for the given collection and OData parameters."""
+        """Search adgs products for the given collection and OData parameters."""
 
         return process_product_search(
             collection.get("station", "adgs"),

services/cadip/rs_server_cadip/api/cadip_search.py

@@ -18,9 +18,12 @@
 It includes an API endpoint, utility functions, and initialization for accessing EODataAccessGateway.
 """
 
-# pylint: disable=redefined-builtin
 import json
+import threading
 import traceback
+
+# pylint: disable=redefined-builtin
+from collections import defaultdict
 from typing import Annotated, List, Literal, Union
 
 import requests
@@ -37,6 +40,7 @@
 from rs_server_cadip.cadip_utils import (
     CADIP_CONFIG,
     cadip_map_mission,
+    cadip_stac_mapper,
     link_assets_to_session,
     prepare_collection,
     read_conf,
@@ -50,6 +54,7 @@
     set_eodag_auth_token,
 )
 from rs_server_common.data_retrieval.provider import CreateProviderFailed
+from rs_server_common.rspy_models import Item
 from rs_server_common.stac_api_common import (
     CollectionType,
     DateTimeType,
@@ -97,15 +102,15 @@ def __init__(self, request: Request | None = None, readwrite: Literal["r", "w"]
         self.sortby = "-published"
 
     @handle_exceptions
-    def process_search(
+    def process_search(  # type: ignore
         self,
         collection: dict,
         odata_params: dict,
         limit: int,
         page: int,
     ) -> stac_pydantic.ItemCollection:
-        """Do the search for the given collection and OData parameters."""
-        session_data = process_session_search(
+        """Search cadip sessions the given collection and OData parameters."""
+        return process_session_search(
             self.request,
             collection.get("station", "cadip"),
             odata_params.get("SessionId", []),
@@ -116,18 +121,30 @@ def process_search(
             limit,
             page,
         )
-        if not session_data.features:
-            # If there are no sessions, don't proceed to assets allocation
-            return session_data
 
-        # To be updated with proper ('')
-        features_ids = ", ".join(feature.id for feature in session_data.features)
+    @handle_exceptions
+    def process_asset_search(
+        self,
+        station: str,
+        session_features: list[Item],
+    ):
+        """
+        Search cadip files for each input cadip session and their associated station.
+        Update input session assets with their associated files.
+
+        Args:
+            station (str): station identifier
+            session_features (list[Item]): sessions as Item objects
+        """
+
+        # Join session ids with ', '
+        features_ids = ", ".join(feature.id for feature in session_features)
 
         assets: list[dict] = []
         page = 1
         while True:
             chunked_assets = process_files_search(
-                collection.get("station", "cadip"),
+                station,
                 features_ids,
                 map_to_session=True,
                 page=page,
@@ -141,8 +158,49 @@ def process_search(
             # If assets are equal to maximum limit, then send another request for the next page
             page += 1
 
-        with open(CADIP_CONFIG / "cadip_stac_mapper.json", encoding="utf-8") as mapper:
-            return link_assets_to_session(session_data, assets, json.loads(mapper.read()))
+        # Update input session items with assets
+        link_assets_to_session(session_features, assets)
+
+    def process_files(self, empty_sessions_data: dict) -> dict:
+        """
+        Search cadip files for each input cadip session. Update the sessions data with their files data.
+
+        Args:
+            empty_sessions_data (dict): dict representation of an ItemCollection
+
+        Returns:
+            dict: updated input dict.
+        """
+
+        # Convert input dict into stac object
+        item_collection = stac_pydantic.ItemCollection.model_validate(empty_sessions_data)
+
+        # Group sessions coming from the same collection. {col1: "item1, item2", col2: "item3" }
+        grouped_sessions = defaultdict(list)
+        for session in item_collection.features:
+            grouped_sessions[session.collection].append(session)
+
+        # Update input session assets with their associated files, in separate threads
+        file_threads = [
+            threading.Thread(
+                target=self.process_asset_search,
+                args=(
+                    self.select_config(collection_id)["station"],
+                    session_features,
+                ),
+            )
+            for collection_id, session_features in grouped_sessions.items()
+        ]
+        for thread in file_threads:
+            thread.start()
+        for thread in file_threads:
+            thread.join()
+
+        # Convert back the stac object into dict.
+        # We implemented some custom Item formating, so we do a back and forth conversion
+        # to apply the formating, then finally return a dict.
+        formatted = [Item.model_validate(feature.model_dump()) for feature in item_collection.features]
+        return stac_pydantic.ItemCollection(features=formatted, type=item_collection.type).model_dump()
 
 
 def auth_validation(request: Request, collection_id: str, access_type: str):
@@ -455,15 +513,12 @@ def process_session_search(  # type: ignore # pylint: disable=too-many-arguments
         products = validate_products(products)
         feature_template_path = CADIP_CONFIG / "cadip_session_ODataToSTAC_template.json"
         stac_mapper_path = CADIP_CONFIG / "cadip_sessions_stac_mapper.json"
-        expanded_session_mapper_path = CADIP_CONFIG / "cadip_stac_mapper.json"
         with (
             open(feature_template_path, encoding="utf-8") as template,
             open(stac_mapper_path, encoding="utf-8") as stac_map,
-            open(expanded_session_mapper_path, encoding="utf-8") as expanded_session_mapper,
         ):
             feature_template = json.loads(template.read())
             stac_mapper = json.loads(stac_map.read())
-            expanded_session_mapper = json.loads(expanded_session_mapper.read())
             collection = create_stac_collection(products, feature_template, stac_mapper)
             return prepare_collection(collection)
 
@@ -574,14 +629,9 @@ def process_files_search(  # pylint: disable=too-many-locals
         if kwargs.get("deprecated", False):
             write_search_products_to_db(CadipDownloadStatus, products)
         feature_template_path = CADIP_CONFIG / "ODataToSTAC_template.json"
-        stac_mapper_path = CADIP_CONFIG / "cadip_stac_mapper.json"
-        with (
-            open(feature_template_path, encoding="utf-8") as template,
-            open(stac_mapper_path, encoding="utf-8") as stac_map,
-        ):
+        with open(feature_template_path, encoding="utf-8") as template:
             feature_template = json.loads(template.read())
-            stac_mapper = json.loads(stac_map.read())
-            cadip_item_collection = create_stac_collection(products, feature_template, stac_mapper)
+            cadip_item_collection = create_stac_collection(products, feature_template, cadip_stac_mapper())
         logger.info("Succesfully listed and processed products from CADIP station")
         if kwargs.get("map_to_session", False):
             return [product.properties for product in products]

services/cadip/rs_server_cadip/cadip_utils.py

@@ -30,6 +30,7 @@
 import eodag
 import yaml
 from fastapi import HTTPException, status
+from rs_server_common.rspy_models import Item
 from rs_server_common.stac_api_common import map_stac_platform
 from rs_server_common.utils.logging import Logging
 from stac_pydantic import ItemCollection, ItemProperties
@@ -48,7 +49,15 @@ def read_conf():
     cadip_search_config = os.environ.get("RSPY_CADIP_SEARCH_CONFIG", str(search_yaml.absolute()))
     with open(cadip_search_config, encoding="utf-8") as search_conf:
         config = yaml.safe_load(search_conf)
-    return config  # WARNING: if the caller wants to modify this cached object, it must deepcopy it first
+    return config  # WARNING: if the caller wants to modify this cached object, he must deepcopy it first
+
+
+@lru_cache()
+def cadip_stac_mapper():
+    """Used each time to read the cadip_stac_mapper config yaml."""
+    with open(CADIP_CONFIG / "cadip_stac_mapper.json", encoding="utf-8") as mapper:
+        config = json.loads(mapper.read())
+    return config  # WARNING: if the caller wants to modify this cached object, he must deepcopy it first
 
 
 def select_config(configuration_id: str) -> dict | None:
@@ -187,14 +196,16 @@ def cadip_reverse_map_mission(platform: Union[str, None]) -> Tuple[Union[str, No
     return None, None
 
 
-def link_assets_to_session(session_data, assets_dict, mapper):
-    """Function used to allocate assets to propper session item based on session id property."""
+def link_assets_to_session(session_features: list[Item], asset_items: list[dict]):
+    """Update input session items with associated assets based on session id property."""
     # Validity check to be later added.
-    for feature in session_data.features:
-        matching_assets = [asset_item for asset_item in assets_dict if feature.id == asset_item["SessionId"]]
+    for feature in session_features:
+        matching_assets = [asset_item for asset_item in asset_items if feature.id == asset_item["SessionId"]]
         for asset_item in matching_assets:
             asset_dict = {
-                map_key: asset_item[map_value] for map_key, map_value in mapper.items() if map_value in asset_item
+                map_key: asset_item[map_value]
+                for map_key, map_value in cadip_stac_mapper().items()
+                if map_value in asset_item
             }
             asset: Asset = Asset(title=asset_dict.pop("id"), roles=["cadu"], **asset_dict)
             if asset.title:
@@ -223,7 +234,6 @@ def link_assets_to_session(session_data, assets_dict, mapper):
         except ValueError as e:
             logger.warning(f"Cannot update start/end datetime for {feature.id}: {e}")
             continue
-    return session_data
 
 
 def prepare_collection(collection: ItemCollection) -> ItemCollection:

services/common/rs_server_common/rspy_models.py

@@ -16,8 +16,10 @@
 Module used to overwrite stac_pydantic with RSPY types.
 """
 
+from datetime import datetime
+
 # mypy: ignore-errors
-from typing import Optional, Sequence
+from typing import Any, Optional, Sequence
 
 import stac_pydantic
 from geojson_pydantic import FeatureCollection
@@ -47,6 +49,16 @@ class ItemProperties(WrapStacCommonMetadata):
 
     model_config = ConfigDict(extra="allow")
 
+    def __init__(self, **data: Any):
+        """Force convert datetime to str if any in init."""
+        data = {
+            key: (value.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" if isinstance(value, datetime) else value)
+            for key, value in data.items()
+        }
+
+        # Call the parent class's initializer
+        super().__init__(**data)
+
 
 class Item(stac_pydantic.item.Item):
     """

services/common/rs_server_common/stac_api_common.py

@@ -570,6 +570,11 @@ def read_cql(filt: dict):
         data = ItemCollection(features=list(all_items.values()), type="FeatureCollection")
         dict_data: Dict[str, Any] = self.paginate(data)
 
+        # In cadip, we retrieved the sessions data.
+        # We need to fill their assets with the session files data.
+        if self.cadip:
+            dict_data = self.process_files(dict_data)
+
         # Handle pagination links.
         if len(dict_data["features"]) > 0:
             # Don't create next page if the current one does not have features
@@ -667,7 +672,7 @@ def process_collection(  # pylint: disable=too-many-locals, too-many-branches, t
             # Don't forward limit value for /search endpoints
             # just use maximum to gather all possible results, page is always 1
             if "/search" in self.request.url.path:
-                search_limit = SEARCH_LIMIT
+                search_limit = self.limit * self.page
                 search_page = 1
 
             # Do the search for this collection
@@ -704,6 +709,25 @@ def process_search(
     ) -> ItemCollection:
         """Do the search for the given collection and OData parameters."""
 
+    def process_asset_search(
+        self,
+        station: str,
+        session_features: list[Item],
+    ):
+        """
+        Implemented only by cadip.
+        Search cadip files for each input cadip session and their associated station.
+        Update input session assets with their associated files.
+        """
+        raise NotImplementedError
+
+    def process_files(self, empty_sessions_data: dict) -> dict:
+        """
+        Implemented only by cadip.
+        Search cadip files for each input cadip session. Update the sessions data with their files data.
+        """
+        raise NotImplementedError
+
 
 def create_collection(collection: dict) -> stac_pydantic.Collection:
     """Used to create stac_pydantic Model Collection based on given collection data."""