cadip_search.md¶

Module for interacting with CADU system through a FastAPI APIRouter.

This module provides functionality to retrieve a list of products from the CADU system for a specified station. It includes an API endpoint, utility functions, and initialization for accessing EODataAccessGateway.

`MockPgstacCadip` ¶

Bases: MockPgstac

Cadip implementation of MockPgstac

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

class MockPgstacCadip(MockPgstac):
    """Cadip implementation of MockPgstac"""

    def __init__(self, request: Request | None = None, readwrite: Literal["r", "w"] | None = None):
        """Constructor"""
        super().__init__(
            request=request,
            readwrite=readwrite,
            service="cadip",
            all_collections=lambda: read_conf()["collections"],
            select_config=select_config,
            stac_to_odata=stac_to_odata,
            map_mission=cadip_map_mission,
            temporal_mapping={
                "start_datetime": "DownlinkStart",
                "datetime": "PublicationDate",
                "published": "PublicationDate",
                "cadip:planned_data_start": "PlannedDataStart",
                "cadip:planned_data_stop": "PlannedDataStop",
                # impossible to define a temporal OData mapping that would be
                # "end_datetime": "max(Files.PublicationDate)"
            },
        )

        # Default sortby value
        self.sortby = "-published"

    @handle_exceptions
    def process_search(  # type: ignore
        self,
        station: str,
        odata_params: dict,
        collection_provider: Callable[[dict], str | None],
        limit: int,
        page: int,
    ) -> ItemCollection:
        """Search cadip sessions for the given station and OData parameters."""
        return process_session_search(station, odata_params, collection_provider, self.sortby, limit, page)

    @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)
        logger.debug(f"Searching for CADIP files at station {station} for session ids: {features_ids}")

        assets: list[dict] = []
        page = 1
        while True:
            chunked_assets = process_files_search(
                station,
                {"SessionId": features_ids},
                map_to_session=True,
                page=page,
            )
            # Gather results into assets list for later allocation
            assets.extend(chunked_assets)

            if len(chunked_assets) < DEFAULT_FILES_LIMIT:
                # If assets are less than maximum limit, then session is complete.
                break
            # If assets are equal to maximum limit, then send another request for the next page
            page += 1

        # Update input session items with assets
        link_assets_to_session(
            session_features,
            sorted(
                assets,
                key=lambda x: dt.fromisoformat(x["PublicationDate"].replace("Z", "+00:00")),
                reverse=False,
            ),
        )

        # Customize sessions
        for feature in session_features:

            # eopf:origin_datetime is required for the PI computing.
            # It has the same value than the end_datetime.
            setattr(feature.properties, "eopf:origin_datetime", feature.properties.end_datetime)

    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 = 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
        run_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()
                if collection_id
            ],
        )

        # 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 ItemCollection(features=formatted, type=item_collection.type).model_dump()

`init(request=None, readwrite=None)` ¶

Constructor

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

def __init__(self, request: Request | None = None, readwrite: Literal["r", "w"] | None = None):
    """Constructor"""
    super().__init__(
        request=request,
        readwrite=readwrite,
        service="cadip",
        all_collections=lambda: read_conf()["collections"],
        select_config=select_config,
        stac_to_odata=stac_to_odata,
        map_mission=cadip_map_mission,
        temporal_mapping={
            "start_datetime": "DownlinkStart",
            "datetime": "PublicationDate",
            "published": "PublicationDate",
            "cadip:planned_data_start": "PlannedDataStart",
            "cadip:planned_data_stop": "PlannedDataStop",
            # impossible to define a temporal OData mapping that would be
            # "end_datetime": "max(Files.PublicationDate)"
        },
    )

    # Default sortby value
    self.sortby = "-published"

`process_asset_search(station, session_features)` ¶

Search cadip files for each input cadip session and their associated station. Update input session assets with their associated files.

Parameters:

Name	Type	Description	Default
`station`	`str`	station identifier	required
`session_features`	`list[Item]`	sessions as Item objects	required

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@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)
    logger.debug(f"Searching for CADIP files at station {station} for session ids: {features_ids}")

    assets: list[dict] = []
    page = 1
    while True:
        chunked_assets = process_files_search(
            station,
            {"SessionId": features_ids},
            map_to_session=True,
            page=page,
        )
        # Gather results into assets list for later allocation
        assets.extend(chunked_assets)

        if len(chunked_assets) < DEFAULT_FILES_LIMIT:
            # If assets are less than maximum limit, then session is complete.
            break
        # If assets are equal to maximum limit, then send another request for the next page
        page += 1

    # Update input session items with assets
    link_assets_to_session(
        session_features,
        sorted(
            assets,
            key=lambda x: dt.fromisoformat(x["PublicationDate"].replace("Z", "+00:00")),
            reverse=False,
        ),
    )

    # Customize sessions
    for feature in session_features:

        # eopf:origin_datetime is required for the PI computing.
        # It has the same value than the end_datetime.
        setattr(feature.properties, "eopf:origin_datetime", feature.properties.end_datetime)

`process_files(empty_sessions_data)` ¶

Search cadip files for each input cadip session. Update the sessions data with their files data.

Parameters:

Name	Type	Description	Default
`empty_sessions_data`	`dict`	dict representation of an ItemCollection	required

Returns:

Name	Type	Description
`dict`	`dict`	updated input dict.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

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 = 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
    run_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()
            if collection_id
        ],
    )

    # 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 ItemCollection(features=formatted, type=item_collection.type).model_dump()

`process_search(station, odata_params, collection_provider, limit, page)` ¶

Search cadip sessions for the given station and OData parameters.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@handle_exceptions
def process_search(  # type: ignore
    self,
    station: str,
    odata_params: dict,
    collection_provider: Callable[[dict], str | None],
    limit: int,
    page: int,
) -> ItemCollection:
    """Search cadip sessions for the given station and OData parameters."""
    return process_session_search(station, odata_params, collection_provider, self.sortby, limit, page)

`auth_validation(request, collection_id, access_type)` ¶

Check if the user KeyCloak roles contain the right for this specific CADIP collection and access type.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The incoming HTTP request used for authentication checks.	required
`collection_id`	`str`	Used to find the CADIP station ("CADIP", "INS", "MPS", "MTI", "NSG", "SGS") from the RSPY_CADIP_SEARCH_CONFIG config yaml file.	required
`access_type`	`str`	The type of access, such as "download" or "read".	required

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

def auth_validation(request: Request, collection_id: str, access_type: str):
    """
    Check if the user KeyCloak roles contain the right for this specific CADIP collection and access type.

    Args:
        request (Request): The incoming HTTP request used for authentication checks.
        collection_id (str): Used to find the CADIP station ("CADIP", "INS", "MPS", "MTI", "NSG", "SGS")
                            from the RSPY_CADIP_SEARCH_CONFIG config yaml file.
        access_type (str): The type of access, such as "download" or "read".
    """

    # Find the collection which id == the input collection_id
    collection = select_config(collection_id)
    if not collection:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Unknown CADIP collection: {collection_id!r}")
    station = collection["station"]

    # Call the authentication function from the authentication module
    authentication.auth_validation("cadip", access_type, request=request, station=station)

`get_allowed_cadip_collections(request)` `async` ¶

Endpoint to retrieve an object containing collections and links that a user is authorized to
access based on their API key.

This endpoint reads the API key from the request to determine the roles associated with the user. Using these roles, it identifies the stations the user can access and filters the available collections accordingly. The endpoint then constructs a JSON, which includes links to the collections that match the allowed stations.

It begins by extracting roles from the request.state.auth_roles and derives the station names the user has access to.
Then, it filters the collections from the configuration to include only those belonging to the allowed stations.
For each filtered collection, a corresponding STAC collection is created with links to detailed session searches.

The final response is a dictionary representation of the STAC catalog, which includes details about the collections the user is allowed to access.

Returns:

Name	Type	Description
`dict`	`dict`	Object containing an array of Collection objects in the Catalog, and Link relations.

Raises:

Type	Description
`HTTPException`	If there are issues with reading configurations or processing session searches.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get("/cadip/collections")
@handle_exceptions
async def get_allowed_cadip_collections(request: Request) -> dict:
    """
        Endpoint to retrieve an object containing collections and links that a user is authorized to
        access based on their API key.

    This endpoint reads the API key from the request to determine the roles associated with the user.
    Using these roles, it identifies the stations the user can access and filters the available collections
    accordingly. The endpoint then constructs a JSON, which includes links to the collections that match the allowed
    stations.

    - It begins by extracting roles from the `request.state.auth_roles` and derives the station names
      the user has access to.
    - Then, it filters the collections from the configuration to include only those belonging to the
      allowed stations.
    - For each filtered collection, a corresponding STAC collection is created with links to detailed
      session searches.

    The final response is a dictionary representation of the STAC catalog, which includes details about
    the collections the user is allowed to access.

    Returns:
        dict: Object containing an array of Collection objects in the Catalog, and Link relations.

    Raises:
        HTTPException: If there are issues with reading configurations or processing session searches.
    """
    logger.info(f"Starting {request.url.path}")
    authentication.auth_validation("cadip", "landing_page", request=request)
    return await request.app.state.pgstac_client.all_collections(request=request)

`get_cadip_collection(request, collection_id)` `async` ¶

Retrieve a STAC-Compliant Collection for a Specific CADIP Station.

This endpoint fetches and returns session data from an external CADIP server, structured as a STAC-compliant Collection. By specifying a collection_id, the client can retrieve a collection of session metadata related to that CADIP station.

Path Parameters:¶

collection_id (string): The unique identifier of the CADIP collection to retrieve.

Response:¶

The response is a STAC Collection object formatted as a dictionary, which contains links to session details. Each session is represented as a link inside the links array, following the STAC specification. These links point to the detailed metadata for each session.

Key Operations:¶

Configuration Lookup: Reads the relevant configuration from RSPY_CADIP_SEARCH_CONFIG.
CADIP Server Request: Sends a request to the CADIP server to retrieve session data.
STAC Formatting: Transforms the session data into a STAC Collection format.
Link Creation: Adds links to session details in the response.

Responses:¶

200 OK: Returns the STAC Collection containing links to session metadata. If multiple collections are available, returns a list of collections.
422 Unprocessable Entity: Returns an error if the STAC Collection cannot be created due to missing or invalid configuration details.

Raises:¶

HTTPException:
422 Unprocessable Entity: If any configuration data is missing, invalid, or causes an error when creating the STAC Collection.

This endpoint is secured by an API key validator, ensuring that only authorized users can retrieve data from the CADIP station.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get("/cadip/collections/{collection_id}")
@handle_exceptions
async def get_cadip_collection(
    request: Request,
    collection_id: Annotated[str, FPath(title="CADIP collection ID.", max_length=100, description="E.G. ins_s1")],
) -> list[dict] | dict | stac_pydantic.Collection:
    """
    Retrieve a STAC-Compliant Collection for a Specific CADIP Station.

    This endpoint fetches and returns session data from an external CADIP server, structured as a STAC-compliant
    Collection. By specifying a `collection_id`, the client can retrieve a collection of session metadata related to
    that CADIP station.

    ### Path Parameters:
    - `collection_id` (string): The unique identifier of the CADIP collection to retrieve.

    ### Response:
    The response is a STAC Collection object formatted as a dictionary, which contains links to session details.
    Each session is represented as a link inside the `links` array, following the STAC specification. These links point
     to the detailed metadata for each session.

    ### Key Operations:
    1. **Configuration Lookup**: Reads the relevant configuration from `RSPY_CADIP_SEARCH_CONFIG`.
    2. **CADIP Server Request**: Sends a request to the CADIP server to retrieve session data.
    3. **STAC Formatting**: Transforms the session data into a STAC Collection format.
    4. **Link Creation**: Adds links to session details in the response.

    ### Responses:
    - **200 OK**: Returns the STAC Collection containing links to session metadata. If multiple collections are
    available, returns a list of collections.
    - **422 Unprocessable Entity**: Returns an error if the STAC Collection cannot be created due to missing or invalid
    configuration details.

    ### Raises:
    - **HTTPException**:
      - **422 Unprocessable Entity**: If any configuration data is missing, invalid, or causes an error when creating
      the STAC Collection.

    This endpoint is secured by an API key validator, ensuring that only authorized users can retrieve data from the
    CADIP station.
    """
    logger.info(f"Starting {request.url.path}")
    auth_validation(request, collection_id, "read")
    return await request.app.state.pgstac_client.get_collection(collection_id, request)

`get_cadip_collection_item_details(request, collection_id, session_id)` `async` ¶

Retrieve Detailed Information for a specific session in a collection.

This endpoint fetches metadata and asset details for a specific session within a collection from the CADIP station. Clients can request session details by providing the collection_id and session_id as path parameters. The session data is retrieved and converted from the original OData format into the STAC format, which provides standardized metadata for spatiotemporal datasets.

Path Parameters:¶

collection_id (string): The unique identifier of the collection from which the session is being retrieved.
session_id (string): The identifier of the specific session within the collection for which details are requested.

Response:¶

Returns a STAC Item containing metadata and asset details about the requested session, including: - Session metadata: Contains important temporal information (e.g., datetime, start_datetime, and end_datetime), the platform (platform), and session-specific details such as externalIds, cadip:num_channels, sat:acquisition_station, cadip:antenna_id, and more. - Satellite information: Includes satellite attributes such as sat:absolute_orbit, cadip:acquisition_id, and status fields like cadip:antenna_status_ok, cadip:front_end_status_ok, and cadip:downlink_status_ok. - Assets: A collection of asset objects associated with the session. Each asset contains: - A unique asset href (link) pointing to the asset resource. - Metadata such as externalIds, cadip:retransfer, cadip:block_number, cadip:channel, created, eviction_datetime, and file:size. - Asset roles, indicating the type of resource (e.g., "cadu"). - Asset title and name.

Responses:¶

200 OK: If the session details are found, returns the STAC Item in JSON format.
404 Not Found: If the session_id is not found within the specified collection.

The endpoint is protected by an API key validator, which requires appropriate access permissions.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get(path="/cadip/collections/{collection_id}/items/{session_id}", response_class=GeoJSONResponse)
@handle_exceptions
async def get_cadip_collection_item_details(
    request: Request,
    collection_id: Annotated[str, FPath(title="CADIP collection ID.", max_length=100, description="E.G. ins_s1")],
    session_id: Annotated[
        str,
        FPath(title="CADIP session ID.", max_length=100, description="E.G. S1A_20231120061537234567"),
    ],
):
    """
    Retrieve Detailed Information for a specific session in a collection.

    This endpoint fetches metadata and asset details for a specific session within a collection from the CADIP station.
    Clients can request session details by providing the `collection_id` and `session_id` as path parameters.
    The session data is retrieved and converted from the original OData format into the STAC format,
    which provides standardized metadata for spatiotemporal datasets.

    ### Path Parameters:
    - `collection_id` (string): The unique identifier of the collection from which the session is being retrieved.
    - `session_id` (string): The identifier of the specific session within the collection for which details are
    requested.

    ### Response:
    Returns a STAC Item containing metadata and asset details about the requested session, including:
    - **Session metadata**: Contains important temporal information (e.g., `datetime`, `start_datetime`, and
    `end_datetime`),
      the platform (`platform`), and session-specific details such as `externalIds`, `cadip:num_channels`,
      `sat:acquisition_station`, `cadip:antenna_id`, and more.
    - **Satellite information**: Includes satellite attributes such as `sat:absolute_orbit`, `cadip:acquisition_id`, and
    status fields like `cadip:antenna_status_ok`, `cadip:front_end_status_ok`, and `cadip:downlink_status_ok`.
    - **Assets**: A collection of asset objects associated with the session. Each asset contains:
      - A unique asset `href` (link) pointing to the asset resource.
      - Metadata such as `externalIds`, `cadip:retransfer`, `cadip:block_number`, `cadip:channel`,
        `created`, `eviction_datetime`, and `file:size`.
      - Asset `roles`, indicating the type of resource (e.g., "cadu").
      - Asset title and name.

    ### Responses:
    - **200 OK**: If the session details are found, returns the STAC Item in JSON format.
    - **404 Not Found**: If the `session_id` is not found within the specified collection.

    The endpoint is protected by an API key validator, which requires appropriate access permissions.
    """
    logger.info(f"Starting {request.url.path}")
    auth_validation(request, collection_id, "read")
    request.state.session_id = session_id  # save for later
    try:
        item = await request.app.state.pgstac_client.get_item(session_id, collection_id, request)
    except HTTPException:  # validation error, just forward it
        raise
    except Exception as exc:  # stac_fastapi.types.errors.NotFoundError
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Cadip session {session_id!r} not found.",
        ) from exc
    return item

`get_cadip_collection_items(request, collection_id, bbox=None, datetime=None, filter_=None, filter_lang='cql2-text', sortby=None, limit=None, page=None)` `async` ¶

Retrieve a List of items for a specific collection.

This endpoint provides access to a list of sessions for a given collection from the CADIP station. By specifying the collection_id in the path, clients can retrieve session metadata in the form of a STAC (SpatioTemporal Asset Catalog) ItemCollection.

Path Parameters:¶

collection_id (string): The unique identifier of the collection from which session data is being requested.

Response:¶

Returns a STAC ItemCollection containing metadata for each session in the specified collection. Each session is represented as a STAC Item, containing key information such as: - Session metadata: Information about the session's time, satellite, and session ID.

Responses:¶

200 OK: If sessions are found, returns the ItemCollection in JSON format.
404 Not Found: If no matching sessions or collection is found.

Errors:¶

500 Internal Server Error: If an error occurs in reading configurations, creating query parameters, or processing the session search.

This endpoint is protected by an API key validator, ensuring appropriate access to the CADIP station.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get(path="/cadip/collections/{collection_id}/items", response_class=GeoJSONResponse)
@handle_exceptions
async def get_cadip_collection_items(
    request: Request,
    collection_id: CollectionType,
    # stac search parameters
    bbox: BBoxType = None,
    datetime: DateTimeType = None,
    filter_: FilterType = None,
    filter_lang: FilterLangType = "cql2-text",
    sortby: SortByType = None,
    limit: LimitType = None,
    page: PageType = None,
):
    """
    Retrieve a List of items for a specific collection.

    This endpoint provides access to a list of sessions for a given collection from the CADIP station.
    By specifying the `collection_id` in the path, clients can retrieve session metadata in the form of a STAC
    (SpatioTemporal Asset Catalog) ItemCollection.

    ### Path Parameters:
    - `collection_id` (string): The unique identifier of the collection from which session data is being requested.

    ### Response:
    Returns a STAC ItemCollection containing metadata for each session in the specified collection.
    Each session is represented as a STAC Item, containing key information such as:
    - **Session metadata**: Information about the session's time, satellite, and session ID.

    ### Responses:
    - **200 OK**: If sessions are found, returns the ItemCollection in JSON format.
    - **404 Not Found**: If no matching sessions or collection is found.

    ### Errors:
    - **500 Internal Server Error**: If an error occurs in reading configurations, creating query parameters, or
    processing the session search.

    This endpoint is protected by an API key validator, ensuring appropriate access to the CADIP station.
    """
    logger.info(f"Starting {request.url.path}")
    auth_validation(request, collection_id, "read")
    return await request.app.state.pgstac_client.item_collection(
        collection_id,
        request,
        bbox=check_bbox_input(bbox),
        datetime=datetime,
        filter_expr=filter_,
        filter_lang=filter_lang,
        sortby=[sortby] if sortby else None,
        limit=limit,
        page=page,
    )

`get_conformance(request)` `async` ¶

Return the STAC/OGC conformance classes implemented by this server.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get("/cadip/conformance")
async def get_conformance(request: Request):
    """Return the STAC/OGC conformance classes implemented by this server."""
    authentication.auth_validation("cadip", "landing_page", request=request)
    return await request.app.state.pgstac_client.conformance()

`get_root_catalog(request)` `async` ¶

Retrieve the RSPY CADIP Search catalog landing page.

This endpoint generates a STAC (SpatioTemporal Asset Catalog) Catalog object that serves as the landing page for the RSPY CADIP service. The catalog includes basic metadata about the service and links to available collections.

The resulting catalog contains: - id: A unique identifier for the catalog, generated as a UUID. - description: A brief description of the catalog. - title: The title of the catalog. - stac_version: The version of the STAC specification to which the catalog conforms. - conformsTo: A list of STAC and OGC API specifications that the catalog conforms to. - links: A link to the /cadip/collections endpoint where users can find available collections.

The stac_version is set to "1.1.0", and the conformsTo field lists the relevant STAC and OGC API specifications that the catalog adheres to. A link to the collections endpoint is added to the catalog's links field, allowing users to discover available collections in the CADIP service.

Parameters: - request: The HTTP request object which includes details about the incoming request.

Returns: - dict: A dictionary representation of the STAC catalog, including metadata and links.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get("/cadip")
async def get_root_catalog(request: Request):
    """
    Retrieve the RSPY CADIP Search catalog landing page.

    This endpoint generates a STAC (SpatioTemporal Asset Catalog) Catalog object that serves as the landing
    page for the RSPY CADIP service. The catalog includes basic metadata about the service and links to
    available collections.

    The resulting catalog contains:
    - `id`: A unique identifier for the catalog, generated as a UUID.
    - `description`: A brief description of the catalog.
    - `title`: The title of the catalog.
    - `stac_version`: The version of the STAC specification to which the catalog conforms.
    - `conformsTo`: A list of STAC and OGC API specifications that the catalog conforms to.
    - `links`: A link to the `/cadip/collections` endpoint where users can find available collections.

    The `stac_version` is set to "1.1.0", and the `conformsTo` field lists the relevant STAC and OGC API
    specifications that the catalog adheres to. A link to the collections endpoint is added to the catalog's
    `links` field, allowing users to discover available collections in the CADIP service.

    Parameters:
    - request: The HTTP request object which includes details about the incoming request.

    Returns:
    - dict: A dictionary representation of the STAC catalog, including metadata and links.
    """
    logger.info(f"Starting {request.url.path}")
    authentication.auth_validation("cadip", "landing_page", request=request)
    return await request.app.state.pgstac_client.landing_page(request=request)

`home()` `async` ¶

Home endpoint. Redirect to the landing page.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@router.get("/", include_in_schema=False)
async def home():
    """Home endpoint. Redirect to the landing page."""
    return RedirectResponse("/cadip")

`process_files_search(station, queryables, limit=DEFAULT_FILES_LIMIT, **kwargs)` ¶

Endpoint to retrieve a list of products from the CADU system for a specified station. Performs a search for products using the CADIP providerand generates a STAC Feature Collection from the products. Args: station (str): CADIP station identifier (e.g., MTI, SGS, MPU, INU). queryables (dict): Query parameters for filtering results. limit (int, optional): Maximum number of products to return. Defaults to DEFAULT_FILES_LIMIT. **kwargs: Additional search parameters such as sortby and page.

Returns:

Type	Description
`list[dict] \| dict`	list[dict] \| dict: - A STAC-compliant Feature Collection of the search results. - If `map_to_session=True`, returns a list of product properties. - If no products are found, returns an empty list.

Raises:

Type	Description
`HTTPException`	If required search parameters (`PublicationDate` or `SessionId`) are missing.
`HTTPException`	If the pagination limit is less than 1.
`HTTPException`	If an invalid station identifier is provided (`CreateProviderFailed`).
`HTTPException`	If a database connection error occurs (`sqlalchemy.exc.OperationalError`).
`HTTPException`	If there is a connection error with the station (`requests.exceptions.ConnectionError`).
`HTTPException`	If a general failure occurs during the process.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

def process_files_search(  # pylint: disable=too-many-locals
    station: str,
    queryables: dict,
    limit: int | None = DEFAULT_FILES_LIMIT,
    **kwargs,
) -> list[dict] | dict:
    """Endpoint to retrieve a list of products from the CADU system for a specified station.
    Performs a search for products using the CADIP providerand generates a STAC Feature Collection from the products.
    Args:
        station (str): CADIP station identifier (e.g., MTI, SGS, MPU, INU).
        queryables (dict): Query parameters for filtering results.
        limit (int, optional): Maximum number of products to return. Defaults to `DEFAULT_FILES_LIMIT`.
        **kwargs: Additional search parameters such as `sortby` and `page`.

    Returns:
        list[dict] | dict:
            - A STAC-compliant Feature Collection of the search results.
            - If `map_to_session=True`, returns a list of product properties.
            - If no products are found, returns an empty list.

    Raises:
        HTTPException: If required search parameters (`PublicationDate` or `SessionId`) are missing.
        HTTPException: If the pagination limit is less than 1.
        HTTPException: If an invalid station identifier is provided (`CreateProviderFailed`).
        HTTPException: If a database connection error occurs (`sqlalchemy.exc.OperationalError`).
        HTTPException: If there is a connection error with the station (`requests.exceptions.ConnectionError`).
        HTTPException: If a general failure occurs during the process.
    """
    query_datetime = queryables.get("PublicationDate")
    session_id = queryables.get("SessionId")
    if not query_datetime and not session_id:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Missing search parameters")

    if session_id:
        queryables["SessionId"] = split_multiple_values(session_id)

    if limit < 1:  # type: ignore
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Pagination cannot be less 0")
    # Init dataretriever / get products / return
    try:
        sort_kwargs = {"sort_by": validate_sort_input(sortby)} if (sortby := kwargs.get("sortby")) else {}
        # The pagination limit is passed separately to the provider search; remove it from queryables
        # to avoid passing the same keyword twice.
        queryables = {k: v for k, v in queryables.items() if k != "limit"}
        # EODAG 4.x doesn't convert `page` to `next_page_token` when `next_page_token_key=skip`,
        # so compute the skip offset directly to ensure correct pagination.
        page_num = kwargs.get("page", 1)
        next_page_token = (page_num - 1) * limit if limit else None
        products = cadip_retriever.init_cadip_provider(station).search(
            **validate(queryables),
            limit=limit,
            next_page_token=next_page_token,
            **sort_kwargs,
        )

        if kwargs.get("map_to_session", False):
            # logger.debug(f"Retrieved products from CADIP station {station}: {products}")
            return [product.properties for product in products]
        cadip_item_collection = create_stac_collection(
            products,
            cadip_odata_to_stac_template(),
            cadip_stac_mapper(),
            external_ids_scheme="cadip",
        )
        logger.debug(f"Retrieved item collection from CADIP station {station}: {cadip_item_collection}")
        return cadip_item_collection.model_dump()
    # pylint: disable=duplicate-code
    except CreateProviderFailed as exception:
        logger.error(f"Failed to create EODAG provider!\n{traceback.format_exc()}")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Bad station identifier: {exception}",
        ) from exception
    # pylint: disable=duplicate-code
    except sqlalchemy.exc.OperationalError as exception:
        logger.error("Failed to connect to database!")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"Database connection error: {exception}",
        ) from exception
    except requests.exceptions.ConnectionError as exception:
        logger.error("Failed to connect to station!")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"Station {station} connection error: {exception}",
        ) from exception
    except Exception as exception:  # pylint: disable=broad-exception-caught
        logger.error("General failure!")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"General failure: {exception}",
        ) from exception

`process_session_search(station, queryables, collection_provider, sortby, limit, page=1)` ¶

Function to process and to retrieve a list of sessions from any CADIP station.

A valid session search request must contain at least a value for either id, platform, or a time interval (start_date and stop_date correctly defined).

Parameters:

Name	Type	Description	Default
`station`	`str`	CADIP station identifier (e.g., MTI, SGS, MPU, INU).	required
`queryables`	`dict`	Lists of queryables applicable to search op.	required
`collection_provider`	`Callable[[dict], str \| None]`	Function that determines STAC collection for a given OData entity	required
`sortby`	`str`	Sort by +/-fieldName (ascending/descending).	required
`limit`	`int`	Maximum number of products to return. Greater than 0, defaults to 100.	required
`page`	`int`	Page number to be displayed, defaults to first one.	`1`

Returns: dict (dict): A STAC Feature Collection of the sessions.

Raises:

Type	Description
`HTTPException(exceptions)`	If search parameters are missing.
`HTTPException(exceptions)`	If there is a JSON mapping error.
`HTTPException(exceptions)`	If there is a value error during mapping.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

@validate_call(config={"arbitrary_types_allowed": True})
def process_session_search(  # type: ignore # pylint: disable=too-many-arguments, too-many-locals, unused-argument
    station: str,
    queryables: dict,
    collection_provider: Callable[[dict], str | None],
    sortby: str,
    limit: Annotated[
        int | None,
        Query(gt=0, default=100, description="Pagination Limit"),
    ],
    page: int | None = 1,
) -> ItemCollection:
    """Function to process and to retrieve a list of sessions from any CADIP station.

    A valid session search request must contain at least a value for either *id*, *platform*, or a time interval
    (*start_date* and *stop_date* correctly defined).

    Args:
        station (str): CADIP station identifier (e.g., MTI, SGS, MPU, INU).
        queryables (dict): Lists of queryables applicable to search op.
        collection_provider (Callable[[dict], str | None]): Function that determines STAC collection
                                                            for a given OData entity
        sortby (str): Sort by +/-fieldName (ascending/descending).
        limit (int, optional): Maximum number of products to return. Greater than 0, defaults to 100.
        page (int): Page number to be displayed, defaults to first one.
    Returns:
        dict (dict): A STAC Feature Collection of the sessions.

    Raises:
        HTTPException (fastapi.exceptions): If search parameters are missing.
        HTTPException (fastapi.exceptions): If there is a JSON mapping error.
        HTTPException (fastapi.exceptions): If there is a value error during mapping.
    """
    try:
        # Get the cadip session provider
        station_session = f"{station}_session"
        session_provider = cadip_retriever.init_cadip_provider(station_session)
        # Authenticate and search sessions
        sort_kwargs = {"sort_by": validate_sort_input(sortby)} if sortby else {}
        # The pagination limit is passed separately to the provider search; remove it from queryables
        # to avoid passing the same keyword twice.
        queryables = {k: v for k, v in queryables.items() if k != "limit"}
        # EODAG 4.x doesn't convert `page` to `next_page_token` when `next_page_token_key=skip`,
        # so compute the skip offset directly to ensure correct pagination.
        next_page_token = (page - 1) * limit if limit and page else None
        products = session_provider.search(
            **validate(queryables),
            sessions_search=True,
            limit=limit,
            next_page_token=next_page_token,
            **sort_kwargs,
        )

        # The station authentication is the same for both the session and assets providers so copy it manually.
        eodag_gateway = session_provider.client  # same for both providers
        providers_config = (
            eodag_gateway._providers.configs  # pylint: disable=protected-access
        )  # use _providers directly to avoid deepcopy in providers property
        plugin_manager = eodag_gateway._plugins_manager  # pylint: disable=protected-access

        # See: eodag/plugins/manager.py::get_auth_plugins
        auth_session = plugin_manager._build_plugin(  # pylint: disable=protected-access
            station_session,
            providers_config[station_session].auth,
            Authentication,
        )
        auth_assets = plugin_manager._build_plugin(  # pylint: disable=protected-access
            station,
            providers_config[station].auth,
            Authentication,
        )

        # Copy parameters between auth plugins, see: eodag/plugins/authentication/token.py
        try:
            auth_assets.token = auth_session.token
            auth_assets.refresh_token = auth_session.refresh_token
            auth_assets.token_expiration = auth_session.token_expiration

        # If anything goes wrong, just log the error. The token will be fetched two times but it's OK.
        except Exception as e:  # pylint: disable=broad-exception-caught
            logger.error(e)

        products = validate_products(products)
        collection = create_stac_collection(
            products,
            cadip_session_odata_to_stac_template(),
            cadip_session_stac_mapper(),
            collection_provider,
            external_ids_scheme="cadip",
        )
        return prepare_collection(collection)

    except ValueError as exception:
        logger.error(exception)
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exception),
        ) from exception
    except Exception as exception:  # pylint: disable=broad-exception-caught
        logger.error(f"General failure! {exception}")
        if isinstance(exception, HTTPException):
            raise
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"General failure: {exception}",
        ) from exception

`validate(queryables)` ¶

Function used to verify / update CADIP-specific queryables before being sent to eodag.

Source code in docs/rs-server/services/cadip/rs_server_cadip/api/cadip_search.py

def validate(queryables: dict):
    """Function used to verify / update CADIP-specific queryables before being sent to eodag."""
    for queryable_name, queryable_data in queryables.items():
        if queryable_name == "PublicationDate":
            queryables[queryable_name] = validate_inputs_format(queryable_data)
        elif isinstance(queryable_data, (str, list)):
            queryables[queryable_name] = validate_str_list(queryable_data)

    return queryables

rs_server_cadip/api/cadip_search.md¶

MockPgstacCadip ¶

__init__(request=None, readwrite=None) ¶

process_asset_search(station, session_features) ¶

process_files(empty_sessions_data) ¶

process_search(station, odata_params, collection_provider, limit, page) ¶

auth_validation(request, collection_id, access_type) ¶

get_allowed_cadip_collections(request) async ¶

get_cadip_collection(request, collection_id) async ¶

Path Parameters:¶

Response:¶

Key Operations:¶

Responses:¶

Raises:¶

get_cadip_collection_item_details(request, collection_id, session_id) async ¶

Path Parameters:¶

Response:¶

Responses:¶

get_cadip_collection_items(request, collection_id, bbox=None, datetime=None, filter_=None, filter_lang='cql2-text', sortby=None, limit=None, page=None) async ¶

Path Parameters:¶

Response:¶

Responses:¶

Errors:¶

get_conformance(request) async ¶

get_root_catalog(request) async ¶

home() async ¶

process_files_search(station, queryables, limit=DEFAULT_FILES_LIMIT, **kwargs) ¶

process_session_search(station, queryables, collection_provider, sortby, limit, page=1) ¶

validate(queryables) ¶

`MockPgstacCadip` ¶

`init(request=None, readwrite=None)` ¶

`process_asset_search(station, session_features)` ¶

`process_files(empty_sessions_data)` ¶

`process_search(station, odata_params, collection_provider, limit, page)` ¶

`auth_validation(request, collection_id, access_type)` ¶

`get_allowed_cadip_collections(request)` `async` ¶

`get_cadip_collection(request, collection_id)` `async` ¶

`get_cadip_collection_item_details(request, collection_id, session_id)` `async` ¶

`get_cadip_collection_items(request, collection_id, bbox=None, datetime=None, filter_=None, filter_lang='cql2-text', sortby=None, limit=None, page=None)` `async` ¶

`get_conformance(request)` `async` ¶

`get_root_catalog(request)` `async` ¶

`home()` `async` ¶

`process_files_search(station, queryables, limit=DEFAULT_FILES_LIMIT, **kwargs)` ¶

`process_session_search(station, queryables, collection_provider, sortby, limit, page=1)` ¶

`validate(queryables)` ¶