Module stac_fastapi.extensions.third_party.bulk_transactions¶

Bulk transactions extension.

Classes¶

AsyncBaseBulkTransactionsClient¶

class AsyncBaseBulkTransactionsClient(

)

BulkTransactionsClient.

Ancestors (in MRO)¶

abc.ABC

Methods¶

bulk_item_insert¶

def bulk_item_insert(
    self,
    items: stac_fastapi.extensions.third_party.bulk_transactions.Items,
    **kwargs
) -> str

Bulk creation of items.

Parameters:

Name	Type	Description	Default
items	None	list of items.	None

Returns:

Type	Description
None	Message indicating the status of the insert.

BaseBulkTransactionsClient¶

class BaseBulkTransactionsClient(

)

BulkTransactionsClient.

Ancestors (in MRO)¶

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

Bulk creation of items.

Parameters:

Name	Type	Description	Default
items	None	list of items.	None
chunk_size	None	number of items processed at a time.	None

Returns:

Type	Description
None	Message indicating the status of the insert.

BulkTransactionExtension¶

class BulkTransactionExtension(
    client: Union[stac_fastapi.extensions.third_party.bulk_transactions.AsyncBaseBulkTransactionsClient, stac_fastapi.extensions.third_party.bulk_transactions.BaseBulkTransactionsClient],
    conformance_classes: List[str] = [],
    schema_href: Union[str, NoneType] = None
)

Bulk Transaction Extension.

Bulk Transaction extension adds the POST /collections/{collection_id}/bulk_items endpoint to the application for efficient bulk insertion of items. The input to this is an object with an attribute "items", that has a value that is an object with a group of attributes that are the ids of each Item, and the value is the Item entity.

Optionally, clients can specify a "method" attribute that is either "insert" or "upsert". If "insert", then the items will be inserted if they do not exist, and an error will be returned if they do. If "upsert", then the items will be inserted if they do not exist, and updated if they do. This defaults to "insert".

{
    "items": {
        "id1": { "type": "Feature", ... },
        "id2": { "type": "Feature", ... },
        "id3": { "type": "Feature", ... }
    },
    "method": "insert"
}

Ancestors (in MRO)¶

stac_fastapi.types.extension.ApiExtension
abc.ABC

Class variables¶

GET

POST

Methods¶

get_request_model¶

def get_request_model(
    self,
    verb: Union[str, NoneType] = 'GET'
) -> Union[pydantic.main.BaseModel, NoneType]

Return the request model for the extension.method.

The model can differ based on HTTP verb

register¶

def register(
    self,
    app: fastapi.applications.FastAPI
) -> None

Register the extension with a FastAPI application.

Parameters:

Name	Type	Description	Default
app	None	target FastAPI application.	None

Returns:

Type	Description
None	None

BulkTransactionMethod¶

class BulkTransactionMethod(
    /,
    *args,
    **kwargs
)

Bulk Transaction Methods.

Ancestors (in MRO)¶

builtins.str
enum.Enum

Class variables¶

INSERT

UPSERT

name

value

Items¶

class Items(
    __pydantic_self__,
    **data: Any
)

A group of STAC Item objects, in the form of a dictionary from Item.id -> Item.

Ancestors (in MRO)¶

pydantic.main.BaseModel
pydantic.utils.Representation

Class variables¶

Config

Static methods¶

construct¶

def construct(
    _fields_set: Union[ForwardRef('SetStr'), NoneType] = None,
    **values: Any
) -> 'Model'

Creates a new model setting dict and fields_set from trusted or pre-validated data.

Default values are respected, but no other validation is performed. Behaves as if Config.extra = 'allow' was set since it adds all passed values

from_orm¶

def from_orm(
    obj: Any
) -> 'Model'

parse_file¶

def parse_file(
    path: Union[str, pathlib.Path],
    *,
    content_type: 'unicode' = None,
    encoding: 'unicode' = 'utf8',
    proto: pydantic.parse.Protocol = None,
    allow_pickle: bool = False
) -> 'Model'

parse_obj¶

def parse_obj(
    obj: Any
) -> 'Model'

parse_raw¶

def parse_raw(
    b: Union[str, bytes],
    *,
    content_type: 'unicode' = None,
    encoding: 'unicode' = 'utf8',
    proto: pydantic.parse.Protocol = None,
    allow_pickle: bool = False
) -> 'Model'

schema¶

def schema(
    by_alias: bool = True,
    ref_template: 'unicode' = '#/definitions/{model}'
) -> 'DictStrAny'

schema_json¶

def schema_json(
    *,
    by_alias: bool = True,
    ref_template: 'unicode' = '#/definitions/{model}',
    **dumps_kwargs: Any
) -> 'unicode'

update_forward_refs¶

def update_forward_refs(
    **localns: Any
) -> None

Try to update ForwardRefs on fields based on this Model, globalns and localns.

validate¶

def validate(
    value: Any
) -> 'Model'

Methods¶

copy¶

def copy(
    self: 'Model',
    *,
    include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    update: Union[ForwardRef('DictStrAny'), NoneType] = None,
    deep: bool = False
) -> 'Model'

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

Name	Type	Description	Default
include	None	fields to include in new model	None
exclude	None	fields to exclude from new model, as with values this takes precedence over include	None
update	None	values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data	None
deep	None	set to `True` to make a deep copy of the model	None

Returns:

Type	Description
None	new model instance

dict¶

def dict(
    self,
    *,
    include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    by_alias: bool = False,
    skip_defaults: Union[bool, NoneType] = None,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False
) -> 'DictStrAny'

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

json¶

def json(
    self,
    *,
    include: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    exclude: Union[ForwardRef('AbstractSetIntStr'), ForwardRef('MappingIntStrAny'), NoneType] = None,
    by_alias: bool = False,
    skip_defaults: Union[bool, NoneType] = None,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    encoder: Union[Callable[[Any], Any], NoneType] = None,
    models_as_dict: bool = True,
    **dumps_kwargs: Any
) -> 'unicode'

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().