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.