Module stac_fastapi.core.base_database_logic¶
Base database logic.
Classes¶
BaseDatabaseLogic¶
python3
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¶
python3
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¶
python3
def apply_bbox_filter(
self,
search: Any,
bbox: list
) -> Any
Apply bounding box filter to the search.
apply_collections_filter¶
python3
def apply_collections_filter(
self,
search: Any,
collection_ids: list[str]
) -> Any
Apply collections filter to the search.
apply_cql2_filter¶
python3
def apply_cql2_filter(
self,
search: Any,
_filter: dict[str, typing.Any] | None
) -> Any
Apply CQL2 filter to the search.
apply_datetime_filter¶
python3
def apply_datetime_filter(
self,
search: Any,
datetime: str | None
) -> Any
Apply datetime filter to the search.
apply_free_text_filter¶
python3
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¶
python3
def apply_ids_filter(
self,
search: Any,
item_ids: list[str]
) -> Any
Apply IDs filter to the search.
apply_intersects_filter¶
python3
def apply_intersects_filter(
self,
search: Any,
intersects: Any
) -> Any
Apply intersects filter to the search.
apply_stacql_filter¶
python3
def apply_stacql_filter(
self,
search: Any,
op: str,
field: str,
value: float
) -> Any
Apply STACQL filter to the search.
async_prep_create_item¶
python3
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¶
python3
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¶
python3
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¶
python3
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¶
python3
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¶
python3
def check_collection_exists(
self,
collection_id: str
) -> None
Check if a collection exists.
create_catalog¶
python3
def create_catalog(
self,
catalog: dict,
refresh: bool = False
) -> None
Create a catalog in the database.
create_catalog_catalog¶
python3
def create_catalog_catalog(
self,
catalog_id: str,
catalog: Any,
request: Any
) -> Any
Create a sub-catalog.
create_catalog_collection¶
python3
def create_catalog_collection(
self,
catalog_id: str,
collection: Any,
request: Any
) -> Any
Create a collection in a catalog.
create_collection¶
python3
def create_collection(
self,
collection: dict,
refresh: bool = False
) -> None
Create a collection in the database.
create_item¶
python3
def create_item(
self,
item: dict,
refresh: bool = False
) -> None
Create an item in the database.
delete_catalog¶
python3
def delete_catalog(
self,
catalog_id: str,
refresh: bool = False
) -> None
Delete a catalog from the database.
delete_collection¶
python3
def delete_collection(
self,
collection_id: str,
refresh: bool = False
) -> None
Delete a collection from the database.
delete_collections¶
python3
def delete_collections(
self
) -> None
Delete all collections.
delete_item¶
python3
def delete_item(
self,
item_id: str,
collection_id: str,
refresh: bool = False
) -> None
Delete an item from the database.
delete_items¶
python3
def delete_items(
self
) -> None
Delete all items.
execute_search¶
python3
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,
ignore_unavailable: bool = True
) -> tuple[typing.Iterable[dict[str, typing.Any]], int | None, str | None]
Execute the search.
find_catalog¶
python3
def find_catalog(
self,
catalog_id: str
) -> dict
Find a catalog in the database.
find_collection¶
python3
def find_collection(
self,
collection_id: str
) -> dict
Find a collection in the database.
get_all_catalogs¶
python3
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¶
python3
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_catalog¶
python3
def get_catalog(
self,
catalog_id: str,
request: Any,
settings: dict,
limit: int = 100
) -> Any
Get a specific catalog.
get_catalog_catalogs¶
python3
def get_catalog_catalogs(
self,
catalog_id: str,
limit: int,
token: str | None,
request: Any = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]
Get sub-catalogs of a catalog.
Returns:
| Type | Description |
|---|---|
| None | Tuple of (catalogs_list, total_count, next_token). |
get_catalog_children¶
python3
def get_catalog_children(
self,
catalog_id: str,
limit: int,
token: str | None,
request: Any = None,
resource_type: str | None = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]
Get children of a catalog.
Returns:
| Type | Description |
|---|---|
| None | Tuple of (children_list, total_count, next_token). |
get_catalog_collection¶
python3
def get_catalog_collection(
self,
catalog_id: str,
collection_id: str,
request: Any
) -> Any
Get a collection from a catalog.
get_catalog_collection_item¶
python3
def get_catalog_collection_item(
self,
catalog_id: str,
collection_id: str,
item_id: str,
request: Any
) -> Any
Get an item from a collection in a catalog.
get_catalog_collection_items¶
python3
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_expr: str | None = None,
filter_lang: str | None = None,
token: str | None = None,
query: str | None = None,
fields: list[str] | None = None
) -> Any
Get items from a collection in a catalog.
get_catalog_collections¶
python3
def get_catalog_collections(
self,
catalog_id: str,
limit: int,
token: str | None,
request: Any = None
) -> tuple[list[dict[str, typing.Any]], int, str | None]
Get collections of a catalog.
Returns:
| Type | Description |
|---|---|
| None | Tuple of (collections_list, total_count, next_token). |
get_items_mapping¶
python3
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¶
python3
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¶
python3
def get_one_item(
self,
collection_id: str,
item_id: str
) -> dict
Retrieve a single item from the database.
get_queryables_mapping¶
python3
def get_queryables_mapping(
self,
collection_id: str = '*'
) -> dict[str, typing.Any]
Retrieve mapping of Queryables for search.
json_patch_collection¶
python3
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¶
python3
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.
make_search¶
python3
def make_search(
self
) -> Any
Create a search instance.
merge_patch_collection¶
python3
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¶
python3
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.
populate_sort¶
python3
def populate_sort(
self,
sortby: list
) -> Any
Populate sort for the search.
update_catalog¶
python3
def update_catalog(
self,
catalog_id: str,
catalog: Any,
request: Any
) -> Any
Update a catalog.
update_collection¶
python3
def update_collection(
self,
collection_id: str,
collection: stac_fastapi.types.stac.Collection,
**kwargs: Any
) -> None
Update a collection in the database.