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¶

create_catalog¶

def create_catalog(
    self,
    catalog: dict,
    refresh: bool = False
) -> None

Create a catalog in the database.

create_collection¶

def create_collection(
    self,
    collection: dict,
    refresh: bool = False
) -> None

Create a collection in the database.

create_item¶

def create_item(
    self,
    item: dict,
    refresh: bool = False
) -> None

Create an item in the database.

delete_catalog¶

def delete_catalog(
    self,
    catalog_id: str,
    refresh: bool = False
) -> None

Delete a catalog from the database.

delete_collection¶

def delete_collection(
    self,
    collection_id: str,
    refresh: bool = False
) -> None

Delete a collection from the database.

delete_item¶

def delete_item(
    self,
    item_id: str,
    collection_id: str,
    refresh: bool = False
) -> None

Delete an item from the database.

find_catalog¶

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

Find a catalog in the database.

find_collection¶

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

Find a collection in the database.

get_all_catalogs¶

def get_all_catalogs(
    self,
    token: str | None,
    limit: int,
    request: Any = None,
    sort: list[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

Returns:

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

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: Union[Tuple[Union[float, int], Union[float, int], Union[float, int], Union[float, int]], Tuple[Union[float, int], Union[float, int], Union[float, int], Union[float, int], Union[float, int], Union[float, int]], NoneType] = 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_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
) -> dict

Retrieve a single item from the database.

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
) -> 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
) -> dict

Patch a item in the database follows RF6902.

merge_patch_collection¶

def merge_patch_collection(
    self,
    collection_id: str,
    collection: dict,
    base_url: str,
    refresh: bool = True
) -> 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
) -> dict

Patch a item in the database follows RF7396.