Module stac_fastapi.elasticsearch.database_logic

Database logic.

Variables

AGGREGATION_MAPPING

COLLECTIONS_INDEX

DEFAULT_SORT

ES_MAX_URL_LENGTH

ITEMS_INDEX_PREFIX

ITEM_INDICES

MAX_LIMIT

logger

Functions

create_collection_index

def create_collection_index(

) -> None

Create the index for a Collection. The settings of the index template will be used implicitly.

Returns:

Type	Description
None	None

create_index_templates

def create_index_templates(

) -> None

Create index templates for the Collection and Item indices.

Returns:

Type	Description
None	None

delete_item_index

def delete_item_index(
    collection_id: str
)

Delete the index for items in a collection.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection whose items index will be deleted.	None

Classes

DatabaseLogic

class DatabaseLogic(
    async_settings: stac_fastapi.elasticsearch.config.AsyncElasticsearchSettings = NOTHING,
    sync_settings: stac_fastapi.elasticsearch.config.ElasticsearchSettings = NOTHING,
    item_serializer: Type[stac_fastapi.core.serializers.ItemSerializer] = <class 'stac_fastapi.core.serializers.ItemSerializer'>,
    collection_serializer: Type[stac_fastapi.core.serializers.CollectionSerializer] = <class 'stac_fastapi.core.serializers.CollectionSerializer'>,
    catalog_serializer: Type[stac_fastapi.core.serializers.CatalogSerializer] = <class 'stac_fastapi.core.serializers.CatalogSerializer'>,
    extensions: list[str] = NOTHING
)

Database logic.

Ancestors (in MRO)

• stac_fastapi.core.base_database_logic.BaseDatabaseLogic
• abc.ABC

Class variables

COLLECTION_FIELD

GEOMETRY_FIELD

PROPERTIES_DATETIME_FIELD

PROPERTIES_END_DATETIME_FIELD

PROPERTIES_START_DATETIME_FIELD

aggregation_mapping

client

sync_client

Static methods

apply_bbox_filter

def apply_bbox_filter(
    search: elasticsearch.dsl._sync.search.Search,
    bbox: list
)

Filter search results based on bounding box.

Parameters:

Name	Type	Description	Default
search	Search	The search object to apply the filter to.	None
bbox	list	The bounding box coordinates, represented as a list of four values [minx, miny, maxx, maxy].	None

Returns:

Type	Description
None	search (Search): The search object with the bounding box filter applied.

apply_collections_filter

def apply_collections_filter(
    search: elasticsearch.dsl._sync.search.Search,
    collection_ids: list[str]
)

Database logic to search a list of STAC collection ids.

apply_datetime_filter

def apply_datetime_filter(
    search: elasticsearch.dsl._sync.search.Search,
    datetime: str | None
) -> tuple[elasticsearch.dsl._sync.search.Search, dict[str, str | None]]

Apply a filter to search on datetime, start_datetime, and end_datetime fields.

Parameters:

Name	Type	Description	Default
search	None	The search object to filter.	None
datetime	None	str	None

Returns:

Type	Description
None	The filtered search object.

apply_free_text_filter

def apply_free_text_filter(
    search: elasticsearch.dsl._sync.search.Search,
    free_text_queries: list[str] | None
)

Create a free text query for Elasticsearch queries.

This method delegates to the shared implementation in apply_free_text_filter_shared.

Parameters:

Name	Type	Description	Default
search	Search	The search object to apply the query to.	None
free_text_queries	list[str]	None	A list of text strings to search for in the properties.

Returns:

Type	Description
Search	The search object with the free text query applied, or the original search object if no free_text_queries were provided.

apply_ids_filter

def apply_ids_filter(
    search: elasticsearch.dsl._sync.search.Search,
    item_ids: list[str]
)

Database logic to search a list of STAC item ids.

apply_intersects_filter

def apply_intersects_filter(
    search: elasticsearch.dsl._sync.search.Search,
    intersects: stac_fastapi.sfeos_helpers.mappings.Geometry
)

Filter search results based on intersecting geometry.

Parameters:

Name	Type	Description	Default
search	Search	The search object to apply the filter to.	None
intersects	Geometry	The intersecting geometry, represented as a GeoJSON-like object.	None

Returns:

Type	Description
None	search (Search): The search object with the intersecting geometry filter applied.

apply_stacql_filter

def apply_stacql_filter(
    search: elasticsearch.dsl._sync.search.Search,
    op: str,
    field: str,
    value: float
)

Filter search results based on a comparison between a field and a value.

Parameters:

Name	Type	Description	Default
search	Search	The search object to apply the filter to.	None
op	str	The comparison operator to use. Can be 'eq' (equal), 'ne'/'neq' (not equal), 'gt' (greater than), 'gte' (greater than or equal), 'lt' (less than), or 'lte' (less than or equal).	None
field	str	The field to perform the comparison on.	None
value	float	The value to compare the field against.	None

Returns:

Type	Description
None	search (Search): The search object with the specified filter applied.

make_search

def make_search(

)

Database logic to create a Search instance.

populate_sort

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

Create a sort configuration for Elasticsearch queries.

This method delegates to the shared implementation in populate_sort_shared.

Parameters:

Name	Type	Description	Default
sortby	list	A list of sort specifications, each containing a field and direction.	None

Returns:

Type	Description
None	dict[str, dict[str, str]]

Methods

aggregate

def aggregate(
    self,
    collection_ids: list[str] | None,
    aggregations: list[str],
    search: elasticsearch.dsl._sync.search.Search,
    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
)

Return aggregations of STAC Items.

apply_cql2_filter

def apply_cql2_filter(
    self,
    search: elasticsearch.dsl._sync.search.Search,
    _filter: dict[str, typing.Any] | None
)

Apply a CQL2 filter to an Elasticsearch Search object.

This method transforms a dictionary representing a CQL2 filter into an Elasticsearch query and applies it to the provided Search object. If the filter is None, the original Search object is returned unmodified.

Parameters:

Name	Type	Description	Default
search	Search	The Elasticsearch Search object to which the filter will be applied.	None
_filter	dict[str, Any]	None	The filter in dictionary form that needs to be applied to the search. The dictionary should follow the structure required by the to_es function which converts it to an Elasticsearch query.

Returns:

Type	Description
Search	The modified Search object with the filter applied if a filter is provided, otherwise the original Search object.

async_prep_create_item

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

Preps an item for insertion into the database.

Parameters:

Name	Type	Description	Default
item	Item	The item to be prepped for insertion.	None
base_url	str	The base URL used to create the item's self URL.	None

Returns:

Type	Description
Item	The prepped item.

bulk_async

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

Perform a bulk insert of items into the database asynchronously.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection to which the items belong.	None
processed_items	list[Item]	A list of Item objects to be inserted into the database.	None
op_type	str	The operation type for the bulk actions. "create" for insert-only (rejects duplicates at ES/OS level), "index" for upsert. Defaults to "create".	None
**kwargs	Any	Additional keyword arguments, including: - refresh (str, optional): Whether to refresh the index after the bulk insert. Can be "true", "false", or "wait_for". Defaults to the value of self.sync_settings.database_refresh. - refresh (bool, optional): Whether to refresh the index after the bulk insert. - raise_on_error (bool, optional): Whether to raise an error if any of the bulk operations fail. Defaults to the value of self.async_settings.raise_on_bulk_error.	None

Returns:

Type	Description
tuple[int, list[dict[str, Any]]]	A tuple containing: - The number of successfully processed actions (success). - A list of errors encountered during the bulk operation (errors).

bulk_async_prep_create_item

def bulk_async_prep_create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str
) -> stac_fastapi.types.stac.Item

Validate and serialize an item for bulk indexing.

This method performs pre-insertion validation and serialization: - Verifying that the collection the item belongs to exists. - Serializing the item into a database-compatible format.

Note: Duplicate item detection is not performed here. It is handled atomically by the database engine via op_type="create" during the bulk indexing phase.

Parameters:

Name	Type	Description	Default
item	Item	The item to be prepared for insertion.	None
base_url	str	The base URL used to construct the item's self URL.	None

Returns:

Type	Description
Item	The prepared item, serialized into a database-compatible format.

Raises:

Type	Description
NotFoundError	If the collection that the item belongs to does not exist in the database.

bulk_sync

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

Perform a bulk insert of items into the database synchronously.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection to which the items belong.	None
processed_items	list[Item]	A list of Item objects to be inserted into the database.	None
op_type	str	The operation type for the bulk actions. "create" for insert-only (rejects duplicates at ES/OS level), "index" for upsert. Defaults to "create".	None
**kwargs	Any	Additional keyword arguments, including: - refresh (str, optional): Whether to refresh the index after the bulk insert. Can be "true", "false", or "wait_for". Defaults to the value of self.sync_settings.database_refresh. - refresh (bool, optional): Whether to refresh the index after the bulk insert. - raise_on_error (bool, optional): Whether to raise an error if any of the bulk operations fail. Defaults to the value of self.sync_settings.raise_on_bulk_error.	None

Returns:

Type	Description
tuple[int, list[dict[str, Any]]]	A tuple containing: - The number of successfully processed actions (success). - A list of errors encountered during the bulk operation (errors).

bulk_sync_prep_create_item

def bulk_sync_prep_create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str
) -> stac_fastapi.types.stac.Item

Validate and serialize an item for bulk indexing.

This method performs pre-insertion validation and serialization: - Verifying that the collection the item belongs to exists. - Serializing the item into a database-compatible format.

Note: Duplicate item detection is not performed here. It is handled atomically by the database engine via op_type="create" during the bulk indexing phase.

Parameters:

Name	Type	Description	Default
item	Item	The item to be prepared for insertion.	None
base_url	str	The base URL used to construct the item's self URL.	None

Returns:

Type	Description
Item	The prepared item, serialized into a database-compatible format.

Raises:

Type	Description
NotFoundError	If the collection that the item belongs to does not exist in the database.

check_collection_exists

def check_collection_exists(
    self,
    collection_id: str
)

Database logic to check if a collection exists.

create_catalog

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

Create a catalog in Elasticsearch.

Parameters:

Name	Type	Description	Default
catalog	dict	The catalog document to create.	None
refresh	bool	Whether to refresh the index after creation.	None

create_catalog_catalog

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

Create a sub-catalog within a catalog.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the parent catalog.	None
catalog	Any	The catalog object to create.	None
request	Any	The request object.	None

Returns:

Type	Description
None	The created catalog.

Raises:

Type	Description
NotFoundError	If the parent catalog does not exist.

create_catalog_collection

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

Create a collection within a catalog.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the parent catalog.	None
collection	Any	The collection object to create.	None
request	Any	The request object.	None

Returns:

Type	Description
None	The created collection.

Raises:

Type	Description
NotFoundError	If the parent catalog does not exist.

create_collection

def create_collection(
    self,
    collection: stac_fastapi.types.stac.Collection,
    **kwargs: Any
)

Create a single collection in the database.

Parameters:

Name	Type	Description	Default
collection	Collection	The Collection object to be created.	None
**kwargs	None	Additional keyword arguments. - refresh (str): Whether to refresh the index after the operation. Can be "true", "false", or "wait_for". - refresh (bool): Whether to refresh the index after the operation. Defaults to the value in self.async_settings.database_refresh.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
ConflictError	If a Collection with the same id already exists in the database.

create_item

def create_item(
    self,
    item: stac_fastapi.types.stac.Item,
    base_url: str = '',
    upsert: bool = False,
    **kwargs: Any
)

Database logic for creating one item.

Parameters:

Name	Type	Description	Default
item	Item	The item to be created.	None
base_url	str	The base URL for the item. Defaults to an empty string.	an empty string
upsert	bool	If False (default), performs an insert-only operation that rejects duplicates (op_type="create"). If True, performs an upsert that overwrites existing items (op_type="index").	None
**kwargs	None	Additional keyword arguments. - refresh (str): Whether to refresh the index after the operation. Can be "true", "false", or "wait_for". - refresh (bool): Whether to refresh the index after the operation. Defaults to the value in self.async_settings.database_refresh.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
ItemAlreadyExistsError	If the item already exists and upsert is False.

delete_catalog

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

Delete a catalog from Elasticsearch.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the catalog to delete.	None
refresh	bool	Whether to refresh the index after deletion.	None

delete_collection

def delete_collection(
    self,
    collection_id: str,
    **kwargs: Any
)

Delete a collection from the database.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection to be deleted.	None
kwargs	Any	Additional keyword arguments, including refresh. - refresh (str): Whether to refresh the index after the operation. Can be "true", "false", or "wait_for". - refresh (bool): Whether to refresh the index after the operation. Defaults to the value in self.async_settings.database_refresh.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
NotFoundError	If the collection with the given collection_id is not found in the database.

delete_collections

def delete_collections(
    self
) -> None

Danger. this is only for tests.

delete_item

def delete_item(
    self,
    item_id: str,
    collection_id: str,
    **kwargs: Any
)

Delete a single item from the database.

Parameters:

Name	Type	Description	Default
item_id	str	The id of the Item to be deleted.	None
collection_id	str	The id of the Collection that the Item belongs to.	None
**kwargs	None	Additional keyword arguments. - refresh (str): Whether to refresh the index after the operation. Can be "true", "false", or "wait_for". - refresh (bool): Whether to refresh the index after the operation. Defaults to the value in self.async_settings.database_refresh.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
NotFoundError	If the Item does not exist in the database.

delete_items

def delete_items(
    self
) -> None

Danger. this is only for tests.

execute_search

def execute_search(
    self,
    search: elasticsearch.dsl._sync.search.Search,
    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 a search query with limit and other optional parameters.

Parameters:

Name	Type	Description	Default
search	Search	The search query to be executed.	None
limit	int	The maximum number of results to be returned.	None
token	str	None	The token used to return the next set of results.
sort	dict[str, dict[str, str]]	None	Specifies how the results should be sorted.
collection_ids	list[str]	None	The collection ids to search.
datetime_search	str	Datetime used for index selection.	None
ignore_unavailable	bool	Whether to ignore unavailable collections. Defaults to True.	True

Returns:

Type	Description
tuple[Iterable[dict[str, Any]], int	None, str

Raises:

Type	Description
NotFoundError	If the collections specified in collection_ids do not exist.

find_catalog

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

Find a catalog in Elasticsearch by ID.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the catalog to find.	None

Returns:

Type	Description
dict	The catalog document.

Raises:

Type	Description
NotFoundError	If the catalog is not found.

find_collection

def find_collection(
    self,
    collection_id: str
) -> stac_fastapi.types.stac.Collection

Find and return a collection from the database.

Parameters:

Name	Type	Description	Default
self	None	The instance of the object calling this function.	None
collection_id	str	The ID of the collection to be found.	None

Returns:

Type	Description
Collection	The found collection, represented as a Collection object.

Raises:

Type	Description
NotFoundError	If the collection with the given collection_id is not found 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 Elasticsearch, 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.

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 from all collections safely.

Generates queryables on-the-fly from each collection's item index mappings using a scroll to prevent memory spikes, and a semaphore to prevent connection pool exhaustion.

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: starlette.requests.Request,
    sort: list[dict[str, typing.Any]] | None = None,
    bbox: list[float] | 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 Elasticsearch, supporting pagination.

Parameters:

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

Returns:

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

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
) -> tuple[list[dict[str, typing.Any]], int | None, str | None]

Get sub-catalogs within a catalog.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the parent catalog.	None
limit	int	Number of results to return.	None
token	str	None	Pagination token.
request	Any	The request object.	None

Returns:

Type	Description
None	Tuple containing list of sub-catalogs, next token, and total count.

get_catalog_children

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 | None, str | None]

Get children of a catalog (both sub-catalogs and collections).

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the parent catalog.	None
limit	int	Number of results to return.	None
token	str	None	Pagination token.
request	Any	The request object.	None
resource_type	str	None	Type of resource to filter by (e.g. "Collection", "Catalog").

Returns:

Type	Description
None	Tuple containing list of children, next token, and total count.

get_catalog_collection

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

Get a specific collection within a catalog.

Validation is now handled securely by CatalogsClient. Just return the collection.

get_catalog_collection_item

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

Get a specific item from a collection within a catalog.

Hierarchy validation is already performed by CatalogsClient.

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_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 for a collection within a catalog.

Hierarchy validation is already performed by CatalogsClient. Skip the redundant DB checks and go straight to the search.

get_catalog_collections

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

Get collections within a catalog.

Parameters:

Name	Type	Description	Default
catalog_id	str	The ID of the parent catalog.	None
limit	int	Number of results to return.	None
token	str	None	Pagination token.
request	Any	The request object.	None

Returns:

Type	Description
None	Tuple containing list of collections, next token, and total count.

get_collections_queryables_mapping

def get_collections_queryables_mapping(
    self
) -> dict

Retrieve mapping of Queryables for collection search.

Used when translating CQL2 filters applied to collection search so that field paths are resolved against the collections index rather than the items index.

Returns:

Type	Description
dict	A dictionary containing the Collection Queryables mappings.

get_items_mapping

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

Get the mapping for the specified collection's items index.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection to get items mapping for.	None

Returns:

Type	Description
dict[str, Any]	The mapping information.

get_items_unique_values

def get_items_unique_values(
    self,
    collection_id: str,
    field_names: Iterable[str],
    *,
    limit: int = 100
) -> 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.

Parameters:

Name	Type	Description	Default
collection_id	str	The id of the Collection that the Item belongs to.	None
item_id	str	The id of the Item.	None

Returns:

Type	Description
None	item (dict): A dictionary containing the source data for the Item.

Raises:

Type	Description
NotFoundError	If the specified Item does not exist in the Collection.

get_queryables_mapping

def get_queryables_mapping(
    self,
    collection_id: str = '*'
) -> dict

Retrieve mapping of Queryables for search.

Args: collection_id (str, optional): The id of the Collection the Queryables belongs to. Defaults to "*".

Returns: dict: A dictionary containing the Queryables mappings.

json_patch_collection

def json_patch_collection(
    self,
    collection_id: str,
    operations: list[typing.Union[stac_fastapi.extensions.transaction.request.PatchAddReplaceTest, stac_fastapi.extensions.transaction.request.PatchMoveCopy, stac_fastapi.extensions.transaction.request.PatchRemove]],
    base_url: str,
    create_nest: bool = False,
    refresh: bool = True
) -> stac_fastapi.types.stac.Collection

Database logic for json patching a collection following RF6902.

Parameters:

Name	Type	Description	Default
collection_id	str	Id of collection to be patched.	None
operations	list	list of operations to run.	None
base_url	str	The base URL used for constructing links.	None
refresh	bool	Refresh the index after performing the operation. Defaults to True.	True

Returns:

Type	Description
None	patched collection.

json_patch_item

def json_patch_item(
    self,
    collection_id: str,
    item_id: str,
    operations: list[typing.Union[stac_fastapi.extensions.transaction.request.PatchAddReplaceTest, stac_fastapi.extensions.transaction.request.PatchMoveCopy, stac_fastapi.extensions.transaction.request.PatchRemove]],
    base_url: str,
    create_nest: bool = False,
    refresh: bool = True
) -> stac_fastapi.types.stac.Item

Database logic for json patching an item following RF6902.

Parameters:

Name	Type	Description	Default
collection_id	str	Collection that item belongs to.	None
item_id	str	Id of item to be patched.	None
operations	list	list of operations to run.	None
base_url	str	The base URL used for constructing URLs for the item.	None
refresh	bool	Refresh the index after performing the operation. Defaults to True.	True

Returns:

Type	Description
None	patched item.

merge_patch_collection

def merge_patch_collection(
    self,
    collection_id: str,
    collection: stac_fastapi.extensions.transaction.request.PartialCollection,
    base_url: str,
    refresh: bool = True
) -> stac_fastapi.types.stac.Collection

Database logic for merge patching a collection following RF7396.

Parameters:

Name	Type	Description	Default
collection_id	str	Id of collection to be patched.	None
collection	PartialCollection	The partial collection to be updated.	None
base_url	None	(str): The base URL used for constructing links.	None
refresh	bool	Refresh the index after performing the operation. Defaults to True.	True

Returns:

Type	Description
None	patched collection.

merge_patch_item

def merge_patch_item(
    self,
    collection_id: str,
    item_id: str,
    item: stac_fastapi.extensions.transaction.request.PartialItem,
    base_url: str,
    refresh: bool = True
) -> stac_fastapi.types.stac.Item

Database logic for merge patching an item following RF7396.

Parameters:

Name	Type	Description	Default
collection_id	str	Collection that item belongs to.	None
item_id	str	Id of item to be patched.	None
item	PartialItem	The partial item to be updated.	None
base_url	None	(str): The base URL used for constructing URLs for the item.	None
refresh	bool	Refresh the index after performing the operation. Defaults to True.	True

Returns:

Type	Description
None	patched item.

update_catalog

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

Update a catalog.

update_collection

def update_collection(
    self,
    collection_id: str,
    collection: stac_fastapi.types.stac.Collection,
    **kwargs: Any
) -> None

Update a collection in the database.

Parameters:

Name	Type	Description	Default
collection_id	str	The ID of the collection to be updated.	None
collection	Collection	The Collection object to be used for the update.	None
**kwargs	None	Additional keyword arguments. - refresh (str): Whether to refresh the index after the operation. Can be "true", "false", or "wait_for". - refresh (bool): Whether to refresh the index after the operation. Defaults to the value in self.async_settings.database_refresh.	None

Returns:

Type	Description
None	None

Raises:

Type	Description
NotFoundError	If the collection with the given collection_id is not found in the database.
ConflictError	If a conflict occurs during the update.