Module stac_fastapi.core.core¶
Core client.
Variables¶
BASE_CONFORMANCE_CLASSES
NumType
STAC_VERSION
logger
Classes¶
BulkTransactionsClient¶
class BulkTransactionsClient(
database: stac_fastapi.core.base_database_logic.BaseDatabaseLogic,
settings: stac_fastapi.core.base_settings.ApiBaseSettings,
session: stac_fastapi.core.session.Session = NOTHING
)
A client for posting bulk transactions to a Postgres database.
Attributes¶
Name | Type | Description | Default |
---|---|---|---|
session | None | An instance of Session to use for database connection. |
None |
database | None | An instance of DatabaseLogic to perform database operations. |
None |
Ancestors (in MRO)¶
- stac_fastapi.extensions.third_party.bulk_transactions.BaseBulkTransactionsClient
- abc.ABC
Methods¶
bulk_item_insert¶
def bulk_item_insert(
self,
items: stac_fastapi.extensions.third_party.bulk_transactions.Items,
chunk_size: Union[int, NoneType] = None,
**kwargs
) -> str
Perform a bulk insertion of items into the database using Elasticsearch.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
items | None | The items to insert. | None |
chunk_size | None | The size of each chunk for bulk processing. | None |
**kwargs | None | Additional keyword arguments, such as request and refresh . |
None |
Returns:
Type | Description |
---|---|
None | A string indicating the number of items successfully added. |
preprocess_item¶
def preprocess_item(
self,
item: stac_fastapi.types.stac.Item,
base_url,
method: stac_fastapi.extensions.third_party.bulk_transactions.BulkTransactionMethod
) -> stac_fastapi.types.stac.Item
Preprocess an item to match the data model.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
item | None | The item to preprocess. | None |
base_url | None | The base URL of the request. | None |
method | None | The bulk transaction method. | None |
Returns:
Type | Description |
---|---|
None | The preprocessed item. |
CoreClient¶
class CoreClient(
database: stac_fastapi.core.base_database_logic.BaseDatabaseLogic,
base_conformance_classes: List[str] = NOTHING,
extensions: List[stac_fastapi.types.extension.ApiExtension] = NOTHING,
session: stac_fastapi.core.session.Session = 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'>,
post_request_model=<class 'stac_fastapi.types.search.BaseSearchPostRequest'>,
stac_version: str = '1.0.0',
landing_page_id: str = 'stac-fastapi',
title: str = 'stac-fastapi',
description: str = 'stac-fastapi'
)
Client for core endpoints defined by the STAC specification.
This class is a implementation of AsyncBaseCoreClient
that implements the core endpoints
defined by the STAC specification. It uses the DatabaseLogic
class to interact with the
database, and ItemSerializer
and CollectionSerializer
to convert between STAC objects and
database records.
Attributes¶
Name | Type | Description | Default |
---|---|---|---|
session | Session | A requests session instance to be used for all HTTP requests. | None |
item_serializer | Type[serializers.ItemSerializer] | A serializer class to be used to convert between STAC items and database records. |
None |
collection_serializer | Type[serializers.CollectionSerializer] | A serializer class to be used to convert between STAC collections and database records. |
None |
database | DatabaseLogic | An instance of the DatabaseLogic class that is used to interactwith the database. |
None |
Ancestors (in MRO)¶
- stac_fastapi.types.core.AsyncBaseCoreClient
- stac_fastapi.types.core.LandingPageMixin
- abc.ABC
Class variables¶
post_request_model
Methods¶
all_collections¶
def all_collections(
self,
**kwargs
) -> stac_fastapi.types.stac.Collections
Read all collections from the database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
**kwargs | None | Keyword arguments from the request. | None |
Returns:
Type | Description |
---|---|
None | A Collections object containing all the collections in the database and links to various resources. |
conformance¶
def conformance(
self,
**kwargs
) -> stac_fastapi.types.stac.Conformance
Conformance classes.
Called with GET /conformance
.
Returns:
Type | Description |
---|---|
None | Conformance classes which the server conforms to. |
conformance_classes¶
def conformance_classes(
self
) -> List[str]
Generate conformance classes by adding extension conformance to base
conformance classes.
extension_is_enabled¶
def extension_is_enabled(
self,
extension: str
) -> bool
Check if an api extension is enabled.
get_collection¶
def get_collection(
self,
collection_id: str,
**kwargs
) -> stac_fastapi.types.stac.Collection
Get a collection from the database by its id.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The id of the collection to retrieve. | None |
kwargs | None | Additional keyword arguments passed to the API call. | None |
Returns:
Type | Description |
---|---|
Collection | A Collection object representing the requested collection. |
Raises:
Type | Description |
---|---|
NotFoundError | If the collection with the given id cannot be found in the database. |
get_item¶
def get_item(
self,
item_id: str,
collection_id: str,
**kwargs
) -> stac_fastapi.types.stac.Item
Get an item from the database based on its id and collection id.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The ID of the collection the item belongs to. | None |
item_id | str | The ID of the item to be retrieved. | None |
Returns:
Type | Description |
---|---|
Item | An Item object representing the requested item. |
Raises:
Type | Description |
---|---|
Exception | If any error occurs while getting the item from the database. |
NotFoundError | If the item does not exist in the specified collection. |
get_search¶
def get_search(
self,
request: starlette.requests.Request,
collections: Union[List[str], NoneType] = None,
ids: Union[List[str], NoneType] = 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,
datetime: Union[datetime.datetime, Tuple[datetime.datetime, datetime.datetime], Tuple[datetime.datetime, NoneType], Tuple[NoneType, datetime.datetime], NoneType] = None,
limit: Union[int, NoneType] = 10,
query: Union[str, NoneType] = None,
token: Union[str, NoneType] = None,
fields: Union[List[str], NoneType] = None,
sortby: Union[str, NoneType] = None,
intersects: Union[str, NoneType] = None,
filter: Union[str, NoneType] = None,
filter_lang: Union[str, NoneType] = None,
**kwargs
) -> stac_fastapi.types.stac.ItemCollection
Get search results from the database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collections | Optional[List[str]] | List of collection IDs to search in. | None |
ids | Optional[List[str]] | List of item IDs to search for. | None |
bbox | Optional[BBox] | Bounding box to search in. | None |
datetime | Optional[DateTimeType] | Filter items based on the datetime field. | None |
limit | Optional[int] | Maximum number of results to return. | None |
query | Optional[str] | Query string to filter the results. | None |
token | Optional[str] | Access token to use when searching the catalog. | None |
fields | Optional[List[str]] | Fields to include or exclude from the results. | None |
sortby | Optional[str] | Sorting options for the results. | None |
intersects | Optional[str] | GeoJSON geometry to search in. | None |
kwargs | None | Additional parameters to be passed to the API. | None |
Returns:
Type | Description |
---|---|
ItemCollection | Collection of Item objects representing the search results. |
Raises:
Type | Description |
---|---|
HTTPException | If any error occurs while searching the catalog. |
item_collection¶
def item_collection(
self,
collection_id: str,
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,
datetime: Union[datetime.datetime, Tuple[datetime.datetime, datetime.datetime], Tuple[datetime.datetime, NoneType], Tuple[NoneType, datetime.datetime], NoneType] = None,
limit: int = 10,
token: str = None,
**kwargs
) -> stac_fastapi.types.stac.ItemCollection
Read items from a specific collection in the database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The identifier of the collection to read items from. | None |
bbox | Optional[BBox] | The bounding box to filter items by. | None |
datetime | Optional[DateTimeType] | The datetime range to filter items by. | None |
limit | int | The maximum number of items to return. The default value is 10. | None |
token | str | A token used for pagination. | None |
request | Request | The incoming request. | None |
Returns:
Type | Description |
---|---|
ItemCollection | An ItemCollection object containing the items from the specified collection that meetthe filter criteria and links to various resources. |
Raises:
Type | Description |
---|---|
HTTPException | If the specified collection is not found. |
Exception | If any error occurs while reading the items from the database. |
landing_page¶
def landing_page(
self,
**kwargs
) -> stac_fastapi.types.stac.LandingPage
Landing page.
Called with GET /
.
Returns:
Type | Description |
---|---|
None | API landing page, serving as an entry point to the API. |
post_search¶
def post_search(
self,
search_request: stac_fastapi.types.search.BaseSearchPostRequest,
request: starlette.requests.Request
) -> stac_fastapi.types.stac.ItemCollection
Perform a POST search on the catalog.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
search_request | BaseSearchPostRequest | Request object that includes the parameters for the search. | None |
kwargs | None | Keyword arguments passed to the function. | None |
Returns:
Type | Description |
---|---|
ItemCollection | A collection of items matching the search criteria. |
Raises:
Type | Description |
---|---|
HTTPException | If there is an error with the cql2_json filter. |
EsAsyncBaseFiltersClient¶
class EsAsyncBaseFiltersClient(
)
Defines a pattern for implementing the STAC filter extension.
Ancestors (in MRO)¶
- stac_fastapi.extensions.core.filter.client.AsyncBaseFiltersClient
- abc.ABC
Methods¶
get_queryables¶
def get_queryables(
self,
collection_id: Union[str, NoneType] = None,
**kwargs
) -> Dict[str, Any]
Get the queryables available for the given collection_id.
If collection_id is None, returns the intersection of all queryables over all collections.
This base implementation returns a blank queryable schema. This is not allowed under OGC CQL but it is allowed by the STAC API Filter Extension
github.com/radiantearth/stac-api-spec/tree/master/fragments/filter#queryables
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The id of the collection to get queryables for. | None |
**kwargs | None | additional keyword arguments | None |
Returns:
Type | Description |
---|---|
Dict[str, Any] | A dictionary containing the queryables for the given collection. |
TransactionsClient¶
class TransactionsClient(
database: stac_fastapi.core.base_database_logic.BaseDatabaseLogic,
settings: stac_fastapi.core.base_settings.ApiBaseSettings,
session: stac_fastapi.core.session.Session = NOTHING
)
Transactions extension specific CRUD operations.
Ancestors (in MRO)¶
- stac_fastapi.types.core.AsyncBaseTransactionsClient
- abc.ABC
Methods¶
create_collection¶
def create_collection(
self,
collection: stac_pydantic.collection.Collection,
**kwargs
) -> stac_fastapi.types.stac.Collection
Create a new collection in the database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection | stac_types.Collection | The collection to be created. | None |
kwargs | None | Additional keyword arguments. | None |
Returns:
Type | Description |
---|---|
stac_types.Collection | The created collection object. |
Raises:
Type | Description |
---|---|
ConflictError | If the collection already exists. |
create_item¶
def create_item(
self,
collection_id: str,
item: Union[stac_pydantic.item.Item, stac_pydantic.item_collection.ItemCollection],
**kwargs
) -> Union[stac_fastapi.types.stac.Item, NoneType]
Create an item in the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The id of the collection to add the item to. | None |
item | stac_types.Item | The item to be added to the collection. | None |
kwargs | None | Additional keyword arguments. | None |
Returns:
Type | Description |
---|---|
stac_types.Item | The created item. |
Raises:
Type | Description |
---|---|
NotFound | If the specified collection is not found in the database. |
ConflictError | If the item in the specified collection already exists. |
delete_collection¶
def delete_collection(
self,
collection_id: str,
**kwargs
) -> Union[stac_fastapi.types.stac.Collection, NoneType]
Delete a collection.
This method deletes an existing collection in the database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The identifier of the collection that contains the item. | None |
kwargs | None | Additional keyword arguments. | None |
Returns:
Type | Description |
---|---|
None | None. |
Raises:
Type | Description |
---|---|
NotFoundError | If the collection doesn't exist. |
delete_item¶
def delete_item(
self,
item_id: str,
collection_id: str,
**kwargs
) -> Union[stac_fastapi.types.stac.Item, NoneType]
Delete an item from a collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
item_id | str | The identifier of the item to delete. | None |
collection_id | str | The identifier of the collection that contains the item. | None |
Returns:
Type | Description |
---|---|
Optional[stac_types.Item] | The deleted item, or None if the item was successfully deleted. |
update_collection¶
def update_collection(
self,
collection_id: str,
collection: stac_pydantic.collection.Collection,
**kwargs
) -> stac_fastapi.types.stac.Collection
Update a collection.
This method updates an existing collection in the database by first finding
the collection by the id given in the keyword argument collection_id
.
If no collection_id
is given the id of the given collection object is used.
If the object and keyword collection ids don't match the sub items
collection id is updated else the items are left unchanged.
The updated collection is then returned.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | None | id of the existing collection to be updated | None |
collection | None | A STAC collection that needs to be updated. | None |
kwargs | None | Additional keyword arguments. | None |
Returns:
Type | Description |
---|---|
None | A STAC collection that has been updated in the database. |
update_item¶
def update_item(
self,
collection_id: str,
item_id: str,
item: stac_pydantic.item.Item,
**kwargs
) -> stac_fastapi.types.stac.Item
Update an item in the collection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
collection_id | str | The ID of the collection the item belongs to. | None |
item_id | str | The ID of the item to be updated. | None |
item | stac_types.Item | The new item data. | None |
kwargs | None | Other optional arguments, including the request object. | None |
Returns:
Type | Description |
---|---|
stac_types.Item | The updated item object. |
Raises:
Type | Description |
---|---|
NotFound | If the specified collection is not found in the database. |