Module stac_fastapi.core.catalogs_client¶

Catalogs client implementation for multi-tenant catalogs extension.

Variables¶

COLLECTIONS_INDEX

logger

Classes¶

CatalogsClient¶

class CatalogsClient(
    database: stac_fastapi.core.base_database_logic.BaseDatabaseLogic,
    catalog_serializer: stac_fastapi.core.serializers.CatalogSerializer = <class 'stac_fastapi.core.serializers.CatalogSerializer'>,
    collection_serializer: stac_fastapi.core.serializers.CollectionSerializer = <class 'stac_fastapi.core.serializers.CollectionSerializer'>,
    item_serializer: stac_fastapi.core.serializers.ItemSerializer = <class 'stac_fastapi.core.serializers.ItemSerializer'>,
    core_client: Any = None
)

Catalogs client implementation for the multi-tenant catalogs extension.

This client implements the AsyncBaseCatalogsClient and AsyncCatalogsSearchClient interfaces and delegates to the database layer for all catalog operations.

Ancestors (in MRO)¶

stac_fastapi_catalogs_extension.client.AsyncBaseCatalogsClient
stac_fastapi_catalogs_extension.client.AsyncCatalogsSearchClient
abc.ABC

Methods¶

catalog_search_get¶

def catalog_search_get(
    self,
    catalog_id: str,
    collections: Optional[List[str]] = None,
    ids: Optional[List[str]] = None,
    bbox: Optional[List[float]] = None,
    intersects: str | None = None,
    datetime: str | None = None,
    limit: int | None = None,
    token: str | None = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.item_collection.ItemCollection | starlette.responses.Response

Search items within a catalog using GET parameters.

Parameters:

Name	Type	Description	Default
catalog_id	None	The catalog ID to search within.	None
collections	None	List of collection IDs to search within.	None
ids	None	List of item IDs to search for.	None
bbox	None	Bounding box to search within.	None
intersects	None	GeoJSON geometry to search within.	None
datetime	None	Datetime range to search within.	None
limit	None	Maximum number of results to return.	None
token	None	Pagination token.	None
request	None	FastAPI request object.	None
**kwargs	None	Search parameters from GET query string (contains extensions like 'filter').	None

Returns:

Type	Description
None	ItemCollection with matching items.

catalog_search_post¶

def catalog_search_post(
    self,
    catalog_id: str,
    search_request: stac_fastapi.types.search.BaseSearchPostRequest,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.item_collection.ItemCollection | starlette.responses.Response

Search items within a catalog and its descendants.

Parameters:

Name	Type	Description	Default
catalog_id	None	The catalog ID to search within.	None
search_request	None	The search request parameters.	None
request	None	FastAPI request object.	None
**kwargs	None	Additional keyword arguments.	None

Returns:

Type	Description
None	ItemCollection with matching items.

Raises:

Type	Description
HTTPException	If requested collections are outside catalog scope.

create_catalog¶

def create_catalog(
    self,
    catalog: stac_pydantic.catalog.Catalog,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.catalog.Catalog | starlette.responses.Response

Create a new catalog.

create_catalog_collection¶

def create_catalog_collection(
    self,
    catalog_id: str,
    collection: stac_pydantic.collection.Collection | stac_fastapi_catalogs_extension.types.ObjectUri,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.collection.Collection | starlette.responses.Response

Create a new collection or link an existing collection to catalog.

create_sub_catalog¶

def create_sub_catalog(
    self,
    catalog_id: str,
    catalog: stac_pydantic.catalog.Catalog | stac_fastapi_catalogs_extension.types.ObjectUri,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.catalog.Catalog | starlette.responses.Response

Create a new catalog or link an existing catalog as a sub-catalog.

delete_catalog¶

def delete_catalog(
    self,
    catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> None

Delete a catalog.

get_all_descendant_collections¶

def get_all_descendant_collections(
    self,
    catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> List[str]

BFS DAG crawl to find all descendant collections using parent_ids field.

Parameters:

Name	Type	Description	Default
catalog_id	None	The root catalog ID to start traversal from.	None
request	None	FastAPI request object.	None
**kwargs	None	Additional keyword arguments.	None

Returns:

Type	Description
None	List of all descendant collection IDs.

Raises:

Type	Description
NotFoundError	If the catalog does not exist.

get_catalog¶

def get_catalog(
    self,
    catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.catalog.Catalog

Get a specific catalog by ID according to the Multi-Tenant specification.

get_catalog_children¶

def get_catalog_children(
    self,
    catalog_id: str,
    limit: int | None = None,
    token: str | None = None,
    type: Optional[Literal['Catalog', 'Collection']] = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_fastapi_catalogs_extension.types.Children | starlette.responses.Response

Get all children (Catalogs and Collections) of a specific catalog.

get_catalog_collection¶

def get_catalog_collection(
    self,
    catalog_id: str,
    collection_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.collection.Collection | starlette.responses.Response

Get a specific collection from a catalog (Scoped Route).

get_catalog_collection_item¶

def get_catalog_collection_item(
    self,
    catalog_id: str,
    collection_id: str,
    item_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.item.Item | starlette.responses.Response

Get a specific item from a collection in a catalog.

get_catalog_collection_items¶

def get_catalog_collection_items(
    self,
    catalog_id: str,
    collection_id: str,
    bbox: list[float] | None = None,
    datetime: str | datetime.datetime | None = None,
    limit: int | None = 10,
    token: str | None = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.item_collection.ItemCollection | starlette.responses.Response

Get items from a collection in a catalog with search support.

get_catalog_collections¶

def get_catalog_collections(
    self,
    catalog_id: str,
    limit: int | None = None,
    token: str | None = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.api.collections.Collections | starlette.responses.Response

Get collections linked from a specific catalog.

get_catalog_conformance¶

def get_catalog_conformance(
    self,
    catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> dict | starlette.responses.Response

Get conformance classes specific to this sub-catalog.

get_catalog_queryables¶

def get_catalog_queryables(
    self,
    catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> dict | starlette.responses.Response

Get queryable fields available for filtering in this sub-catalog.

get_catalogs¶

def get_catalogs(
    self,
    limit: int | None = None,
    token: str | None = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_fastapi_catalogs_extension.types.Catalogs | starlette.responses.Response

Get all catalogs with pagination support.

get_sub_catalogs¶

def get_sub_catalogs(
    self,
    catalog_id: str,
    limit: int | None = None,
    token: str | None = None,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_fastapi_catalogs_extension.types.Catalogs | starlette.responses.Response

Get all sub-catalogs of a specific catalog with pagination.

unlink_catalog_collection¶

def unlink_catalog_collection(
    self,
    catalog_id: str,
    collection_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> None

Unlink a collection from a catalog.

unlink_sub_catalog¶

def unlink_sub_catalog(
    self,
    catalog_id: str,
    sub_catalog_id: str,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> None

Unlink a sub-catalog from its parent.

update_catalog¶

def update_catalog(
    self,
    catalog_id: str,
    catalog: stac_pydantic.catalog.Catalog,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.catalog.Catalog | starlette.responses.Response

Update an existing catalog.

update_catalog_collection¶

def update_catalog_collection(
    self,
    catalog_id: str,
    collection_id: str,
    collection: stac_pydantic.collection.Collection,
    request: starlette.requests.Request | None = None,
    **kwargs
) -> stac_pydantic.collection.Collection | starlette.responses.Response

Update a collection's metadata within a catalog context (Scoped Route).

This method updates collection metadata while ensuring the collection remains linked to the specified catalog. This provides DAG safety by operating within a specific catalog context.

Parameters:

Name	Type	Description	Default
catalog_id	None	The ID of the catalog.	None
collection_id	None	The ID of the collection to update.	None
collection	None	The updated collection data.	None
request	None	FastAPI request object.	None
**kwargs	None	Additional keyword arguments.	None

Returns:

Type	Description
None	The updated collection.

Raises:

Type	Description
NotFoundError	If the catalog or collection is not found, or if the collection is not linked to the catalog.