rs_client/stac/stac_base.md

<< Back to index

StacBase class implementation.

StacBase

Bases: RsClient

Base class for interacting with a STAC (SpatioTemporal Asset Catalog) API using pystac-client.

This class provides methods to retrieve STAC collections, items, queryables, and perform searches.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

class StacBase(RsClient):
    """
    Base class for interacting with a STAC (SpatioTemporal Asset Catalog) API using pystac-client.

    This class provides methods to retrieve STAC collections, items, queryables, and perform searches.
    """

    @handle_api_error
    def __init__(  # pylint: disable=too-many-branches, too-many-arguments, too-many-positional-arguments
        self,
        rs_server_href: str | None,
        rs_server_api_key: str | None = None,
        owner_id: str | None = None,
        logger: logging.Logger | None = None,
        stac_href: str | None = None,  # Flag to enable pystac_client for specific subclasses
        headers: dict[str, str] | None = None,
        parameters: dict[str, Any] | None = None,
        ignore_conformance: bool | None = None,
        modifier: Callable[[Collection | Item | ItemCollection | dict[Any, Any]], None] | None = None,
        request_modifier: Callable[[Request], Request | None] | None = None,
        stac_io: StacApiIO | None = None,
        timeout: Timeout | None = TIMEOUT,
    ):
        """
        Initialize the StacBase instance.

        Args:
            rs_server_href (str | None): URL of the RS server.
            rs_server_api_key (str | None, optional): API key for authentication.
            owner_id (str | None, optional): Owner identifier.
            logger (logging.Logger | None, optional): Logger instance.
            stac_href (str | None): STAC API URL.
            headers (Optional[Dict[str, str]], optional): HTTP headers.
            parameters (Optional[Dict[str, Any]], optional): Additional query parameters.
            ignore_conformance (Optional[bool], optional): Whether to ignore conformance.
            modifier (Callable, optional): Function to modify collection, item, or item collection.
            request_modifier (Optional[Callable[[Request], Union[Request, None]]], optional):
                                                                Function to modify requests.
            stac_io (Optional[StacApiIO], optional): Custom STAC API I/O handler.
            timeout (Optional[Timeout], optional): Request timeout.
        """
        # call RsClient init
        super().__init__(rs_server_href, rs_server_api_key, owner_id, logger)

        # Initialize pystac_client.Client only if required (for CadipClient, AuxipClient, StacClient)
        if not stac_href:
            raise RuntimeError("No stac href provided")
        # pystac_client may throw APIError exception this is handled by the decorator handle_api_error
        self.stac_href = stac_href
        if rs_server_api_key:
            if headers is None:
                headers = {}
            headers[APIKEY_HEADER] = rs_server_api_key
        if stac_io is None:
            stac_io = StacApiIO(  # This is what is done in pystac_client/client.py::from_file
                headers=headers,
                parameters=parameters,
                request_modifier=request_modifier,
                timeout=timeout,
            )
        # Save the OAuth2 authentication cookie in the pystac client cookies
        if self.rs_server_oauth2_cookie:
            stac_io.session.cookies.set("session", self.rs_server_oauth2_cookie)
        self.ps_client = Client.open(
            stac_href,
            headers=headers,
            parameters=parameters,
            ignore_conformance=ignore_conformance,
            modifier=modifier,
            request_modifier=request_modifier,
            stac_io=stac_io,
            timeout=timeout,
        )

    ################################
    # Specific STAC implementation #
    ################################
    @handle_api_error
    def get_landing(self) -> dict:
        """
        Retrieve the STAC API landing page.

        Returns:
            dict: The landing page response.

        Raises:
            RuntimeError: If an API error occurs.
        """

        return self.ps_client.to_dict()

    @handle_api_error
    def get_collections(self) -> Iterator[Collection]:
        """
        Retrieve available STAC collections the user has permission to access.

        Returns:
            Iterator[Collection]: An iterator over available collections.

        Raises:
            RuntimeError: If an API error occurs.
        """

        # Get all the available collections
        return self.ps_client.get_collections()

    def _get_ps_collection(self, collection_id: str) -> Collection | CollectionClient:
        """
        Call pystac_client.Client.get_collection and raise if the collection is missing.

        Since pystac-client 0.9.1, get_collection returns None on a 404 instead of raising an APIError.
        We restore the previous behaviour so that our callers keep getting an exception.
        """
        collection = self.ps_client.get_collection(collection_id)
        if collection is None:
            raise APIError(f"NotFoundError: collection {collection_id!r} does not exist")
        return collection

    @lru_cache
    @handle_api_error
    def get_collection(self, collection_id: str) -> Collection | CollectionClient:
        """
        Retrieve a specific STAC collection by ID.

        Args:
            collection_id (str): The ID of the collection.

        Returns:
            Union[Collection, CollectionClient]: The requested collection.

        Raises:
            RuntimeError: If an API error occurs.
        """
        return self._get_ps_collection(collection_id)

    @handle_api_error
    def get_items(
        self,
        collection_id: str,
        items_ids: list[str] | None = None,
        **query_params: Any,
    ) -> Iterator["Item"]:
        """
        Retrieve items from a collection.

        Args:
            collection_id (str): The ID of the collection.
            items_ids (Union[str, None], optional): Specific item ID(s) to retrieve.
            query_params: Extra query parameters forwarded to the collection /items endpoint
                          (non-standard STAC extension, used by some services).

        Returns:
            Iterator[Item]: An iterator over retrieved items.
        Raises:
            RuntimeError: If an API error occurs.
        """

        # Retrieve the collection
        collection = self._get_ps_collection(collection_id)

        # If non-standard query params are provided, call the /items endpoint manually.
        if query_params:
            params = query_params.copy()
            if items_ids and "ids" not in params:
                params["ids"] = ",".join(items_ids)
            self.logger.info(
                "🔍 Retrieving items from collection '%s' with query params: %s.",
                collection_id,
                params,
            )
            items_link_obj = collection.get_single_link("items")
            if items_link_obj is None:
                raise RuntimeError(f"Collection '{collection_id}' has no 'items' link")
            items_link = items_link_obj.get_href()
            if hasattr(self.ps_client, "_request"):
                # Use protected API to call /items with custom query params (default get_items ignores filters).
                response = self.ps_client._request(  # pylint: disable=protected-access
                    "GET",
                    items_link,
                    params=params,
                )
                # pylint: disable=protected-access
                item_collection = self.ps_client._parse_item_collection(  # type: ignore[attr-defined]
                    response,
                    collection,
                )
                # pylint: enable=protected-access
                return iter(item_collection)
            # Fallback for pystac-client versions without _request
            stac_io = getattr(self.ps_client, "_stac_io", None)
            if stac_io and hasattr(stac_io, "read_json"):
                response_dict = stac_io.read_json(items_link, parameters=params)
                return iter(ItemCollection.from_dict(response_dict).items)
            raise RuntimeError("pystac-client API has changed: cannot perform custom /items request with params.")

        # Retrieve a list of items
        if items_ids:
            self.logger.info(f"🔍 Retrieving specific items from collection '{collection_id}'.")

            try:
                return collection.get_items(*items_ids)

            # Avoid pystac-client fallback through /search ; fetch items one by one.
            # collection.get_items(*ids) internally triggers a search request.
            except (APIError, requests.HTTPError, STACError, ValueError, TypeError) as exc:
                self.logger.debug("❌ Direct retrieval failed, fallback to per-item: %s", exc)

                def iter_items():
                    for item_id in items_ids:
                        try:
                            item = collection.get_item(item_id)
                            if item:
                                yield item
                        except (APIError, requests.HTTPError, STACError, ValueError, TypeError) as new_exc:
                            self.logger.warning("Failed to retrieve item '%s': %s", item_id, new_exc)

                return iter_items()

        # Retrieve all items
        self.logger.info(f"Retrieving all items from collection '{collection_id}'.")
        return collection.get_items()

    @handle_api_error
    def get_item(self, collection_id: str, item_id: str) -> Item | None:
        """
        Retrieve a specific item from a collection.

        Args:
            collection_id (str): The collection ID.
            item_id (str): The item ID.

        Returns:
            Item | None: The retrieved item or None if not found.

        Raises:
            RuntimeError: If an API error occurs.
        """

        # Retrieve the collection
        collection = self._get_ps_collection(collection_id)
        item = collection.get_item(item_id)
        if not item:
            self.logger.error(f"Item with ID '{item_id}' not found in collection '{collection_id}'.")
        return item

    @handle_api_error
    def get_collection_queryables(self, collection_id) -> dict[str, Any]:
        """
        Retrieve queryable fields for a specific collection.

        Args:
            collection_id (str): The collection ID.

        Returns:
            Dict[str, Any]: Dictionary of queryable fields.

        Raises:
            RuntimeError: If an API error occurs.
        """

        return self.ps_client.get_merged_queryables([collection_id])

    def get_queryables(self) -> dict[str, Any]:
        """
        Retrieve queryable fields for all collections in the STAC API. These are the available terms for
        usage when writing filter expressions in /search endpoint for all the collections
        NOTE: the pystac-client library doesn't have a function for this action, so the direct call of
        the endpoint is needed

        Returns:
            Dict[str, Any]: Dictionary of queryable fields.

        Raises:
            RuntimeError: If an exception occurs from the request level.
        """
        try:
            href_queryables = self.stac_href + "queryables"
            response = self.http_session.get(
                href_queryables,
                **self.apikey_headers,
                timeout=TIMEOUT,
            )
        except (requests.exceptions.RequestException, requests.exceptions.Timeout) as e:
            self.logger.exception(f"❌ Could not get the response from the endpoint {href_queryables}: {e}")
            raise RuntimeError(
                f"❌ Could not get the response from the endpoint {href_queryables}",
            ) from e
        if not response.ok:
            raise RuntimeError(f"❌ Could not get queryables from {href_queryables}")
        try:
            json_data = response.json()
            return cast(dict[str, Any], json_data)  # Explicitly cast to Dict[str, Any]
        except ValueError as e:
            raise RuntimeError(f"❌ Invalid JSON response from {href_queryables}") from e

    @handle_api_error
    def search(  # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        **kwargs,
    ) -> ItemCollection | None:
        """
        Perform a STAC search using query parameters.

        Returns:
            ItemCollection | None: Retrieved item collection or None if not found.

        Raises:
            RuntimeError: If an API error occurs.
        """
        kwargs.pop("owner_id", None)
        kwargs["datetime"] = kwargs.pop("timestamp", None)
        stac_filter = kwargs.pop("stac_filter", None)
        if (
            isinstance(stac_filter, dict)
            and stac_filter.get("op", "") == "="
            and len(stac_filter.get("args", [])) == 2
            and stac_filter["args"][0] == {"property": "id"}
        ):
            # Optim/Fix required for EarthDataHub
            kwargs["ids"] = stac_filter["args"][1]
        else:
            kwargs["filter"] = stac_filter
        self.logger.info(f"🔍 Performing STAC search with parameters: {kwargs}")

        try:
            items_search = self.ps_client.search(**kwargs)
            self.logger.info(f"STAC search raw response: {items_search.item_collection()}")
            return items_search.item_collection()
        except NotImplementedError:
            self.logger.exception(
                "❌ The API does not conform to the STAC API Item Search spec"
                "or does not have a link with a 'rel' type of 'search' ",
            )
        return None

__init__(rs_server_href, rs_server_api_key=None, owner_id=None, logger=None, stac_href=None, headers=None, parameters=None, ignore_conformance=None, modifier=None, request_modifier=None, stac_io=None, timeout=TIMEOUT)

Initialize the StacBase instance.

Parameters:

Name	Type	Description	Default
rs_server_href	str | None	URL of the RS server.	required
rs_server_api_key	str | None	API key for authentication.	None
owner_id	str | None	Owner identifier.	None
logger	Logger | None	Logger instance.	None
stac_href	str | None	STAC API URL.	None
headers	Optional[Dict[str, str]]	HTTP headers.	None
parameters	Optional[Dict[str, Any]]	Additional query parameters.	None
ignore_conformance	Optional[bool]	Whether to ignore conformance.	None
modifier	Callable	Function to modify collection, item, or item collection.	None
request_modifier	Optional[Callable[[Request], Union[Request, None]]]	Function to modify requests.	None
stac_io	Optional[StacApiIO]	Custom STAC API I/O handler.	None
timeout	Optional[Timeout]	Request timeout.	TIMEOUT

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def __init__(  # pylint: disable=too-many-branches, too-many-arguments, too-many-positional-arguments
    self,
    rs_server_href: str | None,
    rs_server_api_key: str | None = None,
    owner_id: str | None = None,
    logger: logging.Logger | None = None,
    stac_href: str | None = None,  # Flag to enable pystac_client for specific subclasses
    headers: dict[str, str] | None = None,
    parameters: dict[str, Any] | None = None,
    ignore_conformance: bool | None = None,
    modifier: Callable[[Collection | Item | ItemCollection | dict[Any, Any]], None] | None = None,
    request_modifier: Callable[[Request], Request | None] | None = None,
    stac_io: StacApiIO | None = None,
    timeout: Timeout | None = TIMEOUT,
):
    """
    Initialize the StacBase instance.

    Args:
        rs_server_href (str | None): URL of the RS server.
        rs_server_api_key (str | None, optional): API key for authentication.
        owner_id (str | None, optional): Owner identifier.
        logger (logging.Logger | None, optional): Logger instance.
        stac_href (str | None): STAC API URL.
        headers (Optional[Dict[str, str]], optional): HTTP headers.
        parameters (Optional[Dict[str, Any]], optional): Additional query parameters.
        ignore_conformance (Optional[bool], optional): Whether to ignore conformance.
        modifier (Callable, optional): Function to modify collection, item, or item collection.
        request_modifier (Optional[Callable[[Request], Union[Request, None]]], optional):
                                                            Function to modify requests.
        stac_io (Optional[StacApiIO], optional): Custom STAC API I/O handler.
        timeout (Optional[Timeout], optional): Request timeout.
    """
    # call RsClient init
    super().__init__(rs_server_href, rs_server_api_key, owner_id, logger)

    # Initialize pystac_client.Client only if required (for CadipClient, AuxipClient, StacClient)
    if not stac_href:
        raise RuntimeError("No stac href provided")
    # pystac_client may throw APIError exception this is handled by the decorator handle_api_error
    self.stac_href = stac_href
    if rs_server_api_key:
        if headers is None:
            headers = {}
        headers[APIKEY_HEADER] = rs_server_api_key
    if stac_io is None:
        stac_io = StacApiIO(  # This is what is done in pystac_client/client.py::from_file
            headers=headers,
            parameters=parameters,
            request_modifier=request_modifier,
            timeout=timeout,
        )
    # Save the OAuth2 authentication cookie in the pystac client cookies
    if self.rs_server_oauth2_cookie:
        stac_io.session.cookies.set("session", self.rs_server_oauth2_cookie)
    self.ps_client = Client.open(
        stac_href,
        headers=headers,
        parameters=parameters,
        ignore_conformance=ignore_conformance,
        modifier=modifier,
        request_modifier=request_modifier,
        stac_io=stac_io,
        timeout=timeout,
    )

get_collection(collection_id) cached

Retrieve a specific STAC collection by ID.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection.	required

Returns:

Type	Description
Collection | CollectionClient	Union[Collection, CollectionClient]: The requested collection.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@lru_cache
@handle_api_error
def get_collection(self, collection_id: str) -> Collection | CollectionClient:
    """
    Retrieve a specific STAC collection by ID.

    Args:
        collection_id (str): The ID of the collection.

    Returns:
        Union[Collection, CollectionClient]: The requested collection.

    Raises:
        RuntimeError: If an API error occurs.
    """
    return self._get_ps_collection(collection_id)

get_collection_queryables(collection_id)

Retrieve queryable fields for a specific collection.

Parameters:

Name	Type	Description	Default
collection_id	str	The collection ID.	required

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: Dictionary of queryable fields.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def get_collection_queryables(self, collection_id) -> dict[str, Any]:
    """
    Retrieve queryable fields for a specific collection.

    Args:
        collection_id (str): The collection ID.

    Returns:
        Dict[str, Any]: Dictionary of queryable fields.

    Raises:
        RuntimeError: If an API error occurs.
    """

    return self.ps_client.get_merged_queryables([collection_id])

get_collections()

Retrieve available STAC collections the user has permission to access.

Returns:

Type	Description
Iterator[Collection]	Iterator[Collection]: An iterator over available collections.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def get_collections(self) -> Iterator[Collection]:
    """
    Retrieve available STAC collections the user has permission to access.

    Returns:
        Iterator[Collection]: An iterator over available collections.

    Raises:
        RuntimeError: If an API error occurs.
    """

    # Get all the available collections
    return self.ps_client.get_collections()

get_item(collection_id, item_id)

Retrieve a specific item from a collection.

Parameters:

Name	Type	Description	Default
collection_id	str	The collection ID.	required
item_id	str	The item ID.	required

Returns:

Type	Description
Item | None	Item | None: The retrieved item or None if not found.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def get_item(self, collection_id: str, item_id: str) -> Item | None:
    """
    Retrieve a specific item from a collection.

    Args:
        collection_id (str): The collection ID.
        item_id (str): The item ID.

    Returns:
        Item | None: The retrieved item or None if not found.

    Raises:
        RuntimeError: If an API error occurs.
    """

    # Retrieve the collection
    collection = self._get_ps_collection(collection_id)
    item = collection.get_item(item_id)
    if not item:
        self.logger.error(f"Item with ID '{item_id}' not found in collection '{collection_id}'.")
    return item

get_items(collection_id, items_ids=None, **query_params)

Retrieve items from a collection.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection.	required
items_ids	Union[str, None]	Specific item ID(s) to retrieve.	None
query_params	Any	Extra query parameters forwarded to the collection /items endpoint (non-standard STAC extension, used by some services).	{}

Returns:

Type	Description
Iterator[Item]	Iterator[Item]: An iterator over retrieved items.

Raises: RuntimeError: If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def get_items(
    self,
    collection_id: str,
    items_ids: list[str] | None = None,
    **query_params: Any,
) -> Iterator["Item"]:
    """
    Retrieve items from a collection.

    Args:
        collection_id (str): The ID of the collection.
        items_ids (Union[str, None], optional): Specific item ID(s) to retrieve.
        query_params: Extra query parameters forwarded to the collection /items endpoint
                      (non-standard STAC extension, used by some services).

    Returns:
        Iterator[Item]: An iterator over retrieved items.
    Raises:
        RuntimeError: If an API error occurs.
    """

    # Retrieve the collection
    collection = self._get_ps_collection(collection_id)

    # If non-standard query params are provided, call the /items endpoint manually.
    if query_params:
        params = query_params.copy()
        if items_ids and "ids" not in params:
            params["ids"] = ",".join(items_ids)
        self.logger.info(
            "🔍 Retrieving items from collection '%s' with query params: %s.",
            collection_id,
            params,
        )
        items_link_obj = collection.get_single_link("items")
        if items_link_obj is None:
            raise RuntimeError(f"Collection '{collection_id}' has no 'items' link")
        items_link = items_link_obj.get_href()
        if hasattr(self.ps_client, "_request"):
            # Use protected API to call /items with custom query params (default get_items ignores filters).
            response = self.ps_client._request(  # pylint: disable=protected-access
                "GET",
                items_link,
                params=params,
            )
            # pylint: disable=protected-access
            item_collection = self.ps_client._parse_item_collection(  # type: ignore[attr-defined]
                response,
                collection,
            )
            # pylint: enable=protected-access
            return iter(item_collection)
        # Fallback for pystac-client versions without _request
        stac_io = getattr(self.ps_client, "_stac_io", None)
        if stac_io and hasattr(stac_io, "read_json"):
            response_dict = stac_io.read_json(items_link, parameters=params)
            return iter(ItemCollection.from_dict(response_dict).items)
        raise RuntimeError("pystac-client API has changed: cannot perform custom /items request with params.")

    # Retrieve a list of items
    if items_ids:
        self.logger.info(f"🔍 Retrieving specific items from collection '{collection_id}'.")

        try:
            return collection.get_items(*items_ids)

        # Avoid pystac-client fallback through /search ; fetch items one by one.
        # collection.get_items(*ids) internally triggers a search request.
        except (APIError, requests.HTTPError, STACError, ValueError, TypeError) as exc:
            self.logger.debug("❌ Direct retrieval failed, fallback to per-item: %s", exc)

            def iter_items():
                for item_id in items_ids:
                    try:
                        item = collection.get_item(item_id)
                        if item:
                            yield item
                    except (APIError, requests.HTTPError, STACError, ValueError, TypeError) as new_exc:
                        self.logger.warning("Failed to retrieve item '%s': %s", item_id, new_exc)

            return iter_items()

    # Retrieve all items
    self.logger.info(f"Retrieving all items from collection '{collection_id}'.")
    return collection.get_items()

get_landing()

Retrieve the STAC API landing page.

Returns:

Name	Type	Description
dict	dict	The landing page response.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def get_landing(self) -> dict:
    """
    Retrieve the STAC API landing page.

    Returns:
        dict: The landing page response.

    Raises:
        RuntimeError: If an API error occurs.
    """

    return self.ps_client.to_dict()

get_queryables()

Retrieve queryable fields for all collections in the STAC API. These are the available terms for usage when writing filter expressions in /search endpoint for all the collections NOTE: the pystac-client library doesn't have a function for this action, so the direct call of the endpoint is needed

Returns:

Type	Description
dict[str, Any]	Dict[str, Any]: Dictionary of queryable fields.

Raises:

Type	Description
RuntimeError	If an exception occurs from the request level.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

def get_queryables(self) -> dict[str, Any]:
    """
    Retrieve queryable fields for all collections in the STAC API. These are the available terms for
    usage when writing filter expressions in /search endpoint for all the collections
    NOTE: the pystac-client library doesn't have a function for this action, so the direct call of
    the endpoint is needed

    Returns:
        Dict[str, Any]: Dictionary of queryable fields.

    Raises:
        RuntimeError: If an exception occurs from the request level.
    """
    try:
        href_queryables = self.stac_href + "queryables"
        response = self.http_session.get(
            href_queryables,
            **self.apikey_headers,
            timeout=TIMEOUT,
        )
    except (requests.exceptions.RequestException, requests.exceptions.Timeout) as e:
        self.logger.exception(f"❌ Could not get the response from the endpoint {href_queryables}: {e}")
        raise RuntimeError(
            f"❌ Could not get the response from the endpoint {href_queryables}",
        ) from e
    if not response.ok:
        raise RuntimeError(f"❌ Could not get queryables from {href_queryables}")
    try:
        json_data = response.json()
        return cast(dict[str, Any], json_data)  # Explicitly cast to Dict[str, Any]
    except ValueError as e:
        raise RuntimeError(f"❌ Invalid JSON response from {href_queryables}") from e

search(**kwargs)

Perform a STAC search using query parameters.

Returns:

Type	Description
ItemCollection | None	ItemCollection | None: Retrieved item collection or None if not found.

Raises:

Type	Description
RuntimeError	If an API error occurs.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

@handle_api_error
def search(  # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    **kwargs,
) -> ItemCollection | None:
    """
    Perform a STAC search using query parameters.

    Returns:
        ItemCollection | None: Retrieved item collection or None if not found.

    Raises:
        RuntimeError: If an API error occurs.
    """
    kwargs.pop("owner_id", None)
    kwargs["datetime"] = kwargs.pop("timestamp", None)
    stac_filter = kwargs.pop("stac_filter", None)
    if (
        isinstance(stac_filter, dict)
        and stac_filter.get("op", "") == "="
        and len(stac_filter.get("args", [])) == 2
        and stac_filter["args"][0] == {"property": "id"}
    ):
        # Optim/Fix required for EarthDataHub
        kwargs["ids"] = stac_filter["args"][1]
    else:
        kwargs["filter"] = stac_filter
    self.logger.info(f"🔍 Performing STAC search with parameters: {kwargs}")

    try:
        items_search = self.ps_client.search(**kwargs)
        self.logger.info(f"STAC search raw response: {items_search.item_collection()}")
        return items_search.item_collection()
    except NotImplementedError:
        self.logger.exception(
            "❌ The API does not conform to the STAC API Item Search spec"
            "or does not have a link with a 'rel' type of 'search' ",
        )
    return None

handle_api_error(func)

Decorator to handle APIError exceptions in methods that interact with pystac-client.

This decorator wraps methods of the StacBase class that call self.ps_client, catching APIError exceptions and logging them using the instance's logger.

If the logger attribute is not found or is None, it falls back to printing the error.

Parameters:

Name	Type	Description	Default
func	Callable	The method to be wrapped.	required

Returns:

Name	Type	Description
Callable		The wrapped method that catches and logs APIError exceptions,
		then raises a RuntimeError.

Source code in docs/rs-client-libraries/rs_client/stac/stac_base.py

def handle_api_error(func):
    """
    Decorator to handle APIError exceptions in methods that interact with pystac-client.

    This decorator wraps methods of the StacBase class that call `self.ps_client`,
    catching `APIError` exceptions and logging them using the instance's `logger`.

    If the `logger` attribute is not found or is `None`, it falls back to printing the error.

    Args:
        func (Callable): The method to be wrapped.

    Returns:
        Callable: The wrapped method that catches and logs `APIError` exceptions,
        then raises a RuntimeError.
    """

    @wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            return func(self, *args, **kwargs)
        except APIError as e:
            error_message = f"❌ Pystac client returned exception: {e}"
            if hasattr(self, "logger") and self.logger:
                self.logger.exception(error_message)
            else:
                print(error_message)  # Fallback logging
            raise RuntimeError(error_message) from e

    return wrapper