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_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_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_collection¶

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

Find a collection in the database.

get_all_collections¶

def get_all_collections(
    self,
    token: Optional[str],
    limit: int,
    request: Any = None,
    sort: Optional[List[Dict[str, Any]]] = 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: Optional[List[str]] = None,
    filter: Optional[Dict[str, Any]] = None,
    query: Optional[Dict[str, Dict[str, Any]]] = None,
    datetime: Optional[str] = None
) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[int]]

Retrieve a list of collections 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
bbox	Optional[BBox]	Bounding box to filter collections by spatial extent. Defaults to None.	None
q	Optional[List[str]]	Free text search terms. Defaults to None.	None
filter	Optional[Dict[str, Any]]	Structured query in CQL2 format. Defaults to None.	None
query	Optional[Dict[str, Dict[str, Any]]]	Query extension parameters. Defaults to None.	None
datetime	Optional[str]	Temporal filter. Defaults to None.	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, 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.

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.