Module stac_fastapi.core.base_database_logic

Base database logic.

Classes

BaseDatabaseLogic

class BaseDatabaseLogic(
    /,
    *args,
    **kwargs
)

Abstract base class for database logic.

This class defines the basic structure and operations for database interactions. Subclasses must provide implementations for these methods.

Ancestors (in MRO)

• abc.ABC

Descendants

• stac_fastapi.elasticsearch.database_logic.DatabaseLogic
• stac_fastapi.opensearch.database_logic.DatabaseLogic

Methods

aggregate

def aggregate(
    self,
    collection_ids: list[str] | None,
    aggregations: list[str],
    search: Any,
    centroid_geohash_grid_precision: int,
    centroid_geohex_grid_precision: int,
    centroid_geotile_grid_precision: int,
    geometry_geohash_grid_precision: int,
    geometry_geotile_grid_precision: int,
    datetime_frequency_interval: str,
    datetime_search: str,
    ignore_unavailable: bool | None = True
) -> Any

Return aggregations of STAC Items.

apply_bbox_filter

def apply_bbox_filter(
    self,
    search: Any,
    bbox: list
) -> Any

Apply bounding box filter to the search.

apply_collections_filter

def apply_collections_filter(
    self,
    search: Any,
    collection_ids: list[str]
) -> Any

Apply collections filter to the search.

apply_cql2_filter

def apply_cql2_filter(
    self,
    search: Any,
    _filter: dict[str, typing.Any] | None
) -> Any

Apply CQL2 filter to the search.

apply_datetime_filter

def apply_datetime_filter(
    self,
    search: Any,
    datetime: str | None
) -> Any

Apply datetime filter to the search.

apply_free_text_filter

def apply_free_text_filter(
    self,
    search: Any,
    free_text_queries: list[str] | None
) -> Any

Apply free text filter to the search.

apply_ids_filter

def apply_ids_filter(
    self,
    search: Any,
    item_ids: list[str]
) -> Any

Apply IDs filter to the search.

apply_intersects_filter

def apply_intersects_filter(
    self,
    search: Any,
    intersects: Any
) -> Any

Apply intersects filter to the search.

apply_stacql_filter

def apply_stacql_filter(
    self,
    search: Any,
    op: str,
    field: str,
    value: float
) -> Any

Apply STACQL filter to the search.

async_prep_create_item

def async_prep_create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str,
    exist_ok: bool = False
) -> stac_fastapi.types.stac.Item

Prep an item for insertion into the database.

bulk_async

def bulk_async(
    self,
    collection_id: str,
    processed_items: list[stac_fastapi.types.stac.Item],
    **kwargs: Any
) -> tuple[int, list[dict[str, typing.Any]]]

Perform a bulk insert of items into the database asynchronously.

bulk_async_prep_create_item

def bulk_async_prep_create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str,
    exist_ok: bool = False
) -> stac_fastapi.types.stac.Item | None

Prepare an item for bulk insertion.

bulk_sync

def bulk_sync(
    self,
    collection_id: str,
    processed_items: list[stac_fastapi.types.stac.Item],
    **kwargs: Any
) -> tuple[int, list[dict[str, typing.Any]]]

Perform a bulk insert of items into the database synchronously.

bulk_sync_prep_create_item

def bulk_sync_prep_create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str,
    exist_ok: bool = False
) -> stac_fastapi.types.stac.Item | None

Prepare an item for insertion into the database.

check_collection_exists

def check_collection_exists(
    self,
    collection_id: str
) -> None

Check if a collection exists.

check_collection_exists_sync

def check_collection_exists_sync(
    self,
    collection_id: str
) -> None

Check if a collection exists, using the synchronous client.

create_catalog

def create_catalog(
    self,
    catalog: dict,
    refresh: bool = False
) -> stac_fastapi.core.etag.DocumentVersion | None

Create a catalog; reject an id already used by any document.

Returns the version of the stored document.

create_catalog_catalog

def create_catalog_catalog(
    self,
    catalog_id: str,
    catalog: Any,
    request: Any
) -> Any

Create a sub-catalog.

create_catalog_collection

def create_catalog_collection(
    self,
    catalog_id: str,
    collection: Any,
    request: Any
) -> Any

Create a collection in a catalog.

create_collection

def create_collection(
    self,
    collection: dict,
    refresh: bool = False
) -> stac_fastapi.core.etag.DocumentVersion | None

Create a collection in the database; return the version it stored.

create_item

def create_item(
    self,
    item: dict,
    refresh: bool = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None,
    **kwargs: Any
) -> stac_fastapi.core.etag.DocumentVersion | None

Create an item in the database; return the version it stored.

With version (If-Match), only that stored version is replaced; 412 otherwise.

delete_catalog

def delete_catalog(
    self,
    catalog_id: str,
    refresh: bool = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> None

Delete a catalog from the database.

Implementations MUST verify the document exists and is a Catalog before deleting, raising NotFoundError otherwise; deleting a catalog must never remove Collection or Item data. With version (If-Match), only that stored version is deleted, and nothing changes otherwise.

delete_collection

def delete_collection(
    self,
    collection_id: str,
    refresh: bool = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> None

Delete a collection from the database.

delete_collections

def delete_collections(
    self
) -> None

Delete all collections.

delete_item

def delete_item(
    self,
    item_id: str,
    collection_id: str,
    refresh: bool = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> None

Delete an item from the database.

delete_items

def delete_items(
    self
) -> None

Delete all items.

execute_search

def execute_search(
    self,
    search: Any,
    limit: int,
    token: str | None,
    sort: dict[str, dict[str, str]] | None,
    collection_ids: list[str] | None,
    datetime_search: str,
    cql2_metadata: dict[str, typing.Any] | None = None,
    ignore_unavailable: bool = True
) -> tuple[typing.Iterable[dict[str, typing.Any]], int | None, str | None]

Execute the search.

find_catalog

def find_catalog(
    self,
    catalog_id: str
) -> dict

Find a catalog in the database.

find_catalog_with_version

def find_catalog_with_version(
    self,
    catalog_id: str
) -> tuple[dict, stac_fastapi.core.etag.DocumentVersion]

Find a catalog and the version of its stored document.

find_collection

def find_collection(
    self,
    collection_id: str
) -> dict

Find a collection in the database.

find_collection_with_version

def find_collection_with_version(
    self,
    collection_id: str
) -> tuple[dict, stac_fastapi.core.etag.DocumentVersion]

Find a collection and the version of its stored document.

get_all_catalogs

def get_all_catalogs(
    self,
    token: str | None,
    limit: int,
    request: Any = None,
    sort: list[dict[str, typing.Any]] | None = None,
    filter: dict[str, typing.Any] | None = None
) -> tuple[list[dict[str, typing.Any]], str | None, int | None]

Retrieve a list of catalogs from the database, supporting pagination.

Parameters:

Name	Type	Description	Default
token	Optional[str]	The pagination token.	None
limit	int	The number of results to return.	None
request	Any	The FastAPI request object. Defaults to None.	None
sort	Optional[List[Dict[str, Any]]]	Optional sort parameter. Defaults to None.	None
filter	Optional[Dict[str, Any]]	A CQL2 JSON filter the catalogs must match. Defaults to None.	None

Returns:

Type	Description
None	A tuple of (catalogs, next pagination token if any, optional count).

get_all_collection_queryables

def get_all_collection_queryables(
    self
) -> list[dict]

Retrieve all queryables schemas from all collections.

Returns:

Type	Description
None	A list of queryables dictionaries, one from each active collection.

get_all_collections

def get_all_collections(
    self,
    token: str | None,
    limit: int,
    request: Any = None,
    sort: list[dict[str, typing.Any]] | None = None,
    bbox: tuple[float | int, float | int, float | int, float | int] | tuple[float | int, float | int, float | int, float | int, float | int, float | int] | None = None,
    q: list[str] | None = None,
    filter: dict[str, typing.Any] | None = None,
    query: dict[str, dict[str, typing.Any]] | None = None,
    datetime: str | None = None
) -> tuple[list[dict[str, typing.Any]], str | None, int | None]

Retrieve a list of collections from the database, supporting pagination.

Parameters:

Name	Type	Description	Default
token	str	None	The pagination token.
limit	int	The number of results to return.	None
request	Any	The FastAPI request object. Defaults to None.	None
sort	list[dict[str, Any]]	None	Optional sort parameter. Defaults to None.
bbox	BBox	None	Bounding box to filter collections by spatial extent. Defaults to None.
q	list[str]	None	Free text search terms. Defaults to None.
filter	dict[str, Any]	None	Structured query in CQL2 format. Defaults to None.
query	dict[str, dict[str, Any]]	None	Query extension parameters. Defaults to None.
datetime	str	None	Temporal filter. Defaults to None.

Returns:

Type	Description
None	A tuple of (collections, next pagination token if any, optional count).

get_catalog

def get_catalog(
    self,
    catalog_id: str,
    request: Any,
    settings: dict,
    limit: int = 100
) -> Any

Get a specific catalog.

get_catalog_catalogs

def get_catalog_catalogs(
    self,
    catalog_id: str,
    limit: int,
    token: str | None,
    request: Any = None,
    filter: dict[str, typing.Any] | None = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]

Get sub-catalogs of a catalog, optionally matching a CQL2 JSON filter.

Returns:

Type	Description
None	Tuple of (catalogs_list, total_count, next_token).

get_catalog_children

def get_catalog_children(
    self,
    catalog_id: str,
    limit: int,
    token: str | None,
    request: Any = None,
    resource_type: str | None = None,
    filter: dict[str, typing.Any] | None = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]

Get children of a catalog, optionally matching a CQL2 JSON filter.

Returns:

Type	Description
None	Tuple of (children_list, total_count, next_token).

get_catalog_collection

def get_catalog_collection(
    self,
    catalog_id: str,
    collection_id: str,
    request: Any
) -> Any

Get a collection from a catalog.

get_catalog_collection_items

def get_catalog_collection_items(
    self,
    catalog_id: str,
    collection_id: str,
    request: Any,
    bbox: list[float] | None = None,
    datetime: str | None = None,
    limit: int = 10,
    sortby: str | None = None,
    filter: dict[str, typing.Any] | None = None,
    token: str | None = None,
    query: str | None = None,
    fields: list[str] | None = None
) -> Any

Get items from a collection in a catalog, optionally matching a CQL2 JSON filter.

get_catalog_collections

def get_catalog_collections(
    self,
    catalog_id: str,
    limit: int,
    token: str | None,
    request: Any = None,
    filter: dict[str, typing.Any] | None = None,
    sort: list[dict[str, typing.Any]] | None = None,
    q: list[str] | None = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]

Get collections of a catalog.

Optionally matching a CQL2 JSON filter, ordered by sort ({field, direction} pairs), and matching q free-text terms.

Returns:

Type	Description
None	Tuple of (collections_list, total_count, next_token).

get_item_for_write

def get_item_for_write(
    self,
    collection_id: str,
    item_id: str
) -> dict

Retrieve an item for a write without applying read visibility filters.

Under datetime index filtering the real-time mget lookup is used for the same reason as in :meth:get_item_hit_for_write.

get_item_hit_for_write

def get_item_hit_for_write(
    self,
    collection_id: str,
    item_id: str
) -> dict

Retrieve the write-lookup hit for an item, including its version.

Returns a hit dict carrying _source, _index, _seq_no and _primary_term, so callers that then write via create_item can pass the result back as existing_hit (the existence check is not repeated) or read the stored version from it.

Under datetime index filtering, _id uniqueness is enforced per physical index and a just-written document may not be visible to _search yet, so the lookup is a real-time mget over the collection alias's concrete indexes instead.

Raises:

Type	Description
NotFoundError	If the item does not exist in the collection.

get_items_mapping

def get_items_mapping(
    self,
    collection_id: str
) -> dict[str, dict[str, typing.Any]]

Get the mapping for the items in the collection.

get_items_unique_values

def get_items_unique_values(
    self,
    collection_id: str,
    field_names: Iterable[str],
    *,
    limit: int = Ellipsis
) -> dict[str, list[str]]

Get the unique values for the given fields in the collection.

get_one_item

def get_one_item(
    self,
    collection_id: str,
    item_id: str,
    *,
    include_hidden: bool = False
) -> dict

Retrieve a single item from the database.

get_one_item_with_version

def get_one_item_with_version(
    self,
    collection_id: str,
    item_id: str,
    *,
    include_hidden: bool = False
) -> tuple[dict, stac_fastapi.core.etag.DocumentVersion]

Retrieve a single item and the version of its stored document.

With include_hidden, HIDE_ITEM_PATH does not apply, as for a write.

get_queryables_mapping

def get_queryables_mapping(
    self,
    collection_id: str = '*'
) -> dict[str, typing.Any]

Retrieve mapping of Queryables for search.

json_patch_collection

def json_patch_collection(
    self,
    collection_id: str,
    operations: list,
    base_url: str,
    create_nest: bool = False,
    refresh: bool = True,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> dict

Patch a collection in the database follows RF6902.

json_patch_item

def json_patch_item(
    self,
    collection_id: str,
    item_id: str,
    operations: list,
    base_url: str,
    create_nest: bool = False,
    refresh: bool = True,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> dict

Patch a item in the database follows RF6902.

make_search

def make_search(
    self
) -> Any

Create a search instance.

merge_patch_collection

def merge_patch_collection(
    self,
    collection_id: str,
    collection: dict,
    base_url: str,
    refresh: bool = True,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> dict

Patch a collection in the database follows RF7396.

merge_patch_item

def merge_patch_item(
    self,
    collection_id: str,
    item_id: str,
    item: dict,
    base_url: str,
    refresh: bool = True,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> dict

Patch a item in the database follows RF7396.

populate_sort

def populate_sort(
    self,
    sortby: list,
    field_types: dict[str, str] | None = None
) -> Any

Populate sort for the search.

unlink_catalog_orphans

def unlink_catalog_orphans(
    self,
    catalog_id: str
) -> None

Unlink children that still reference a catalog that no longer exists.

Nothing is deleted. A retry of a catalog delete whose cleanup failed calls this, with or without If-Match.

update_catalog

def update_catalog(
    self,
    catalog_id: str,
    catalog: dict,
    refresh: bool | str = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None
) -> stac_fastapi.core.etag.DocumentVersion | None

Replace a catalog, keeping its stored parent_ids; return the version stored.

With version (If-Match), only that stored version is replaced; 412 otherwise.

update_collection

def update_collection(
    self,
    collection_id: str,
    collection: stac_fastapi.types.stac.Collection,
    preserve_parent_ids: bool = False,
    version: stac_fastapi.core.etag.DocumentVersion | None = None,
    **kwargs: Any
) -> stac_fastapi.core.etag.DocumentVersion | None

Update a collection in the database.

update_parent_ids

def update_parent_ids(
    self,
    doc_id: str,
    doc_type: str,
    parent_id: str,
    add: bool,
    refresh: bool | str = False
) -> dict

Atomically add or remove one parent id in a Collection's or Catalog's parent_ids.

Returns:

Type	Description
None	The stored document after the update.