transaction

stac_fastapi.extensions.core.transaction

transaction extension module.

AsyncBaseTransactionsClient

Bases: ABC

Defines a pattern for implementing the STAC transaction extension.

Source code in stac_fastapi/extensions/core/transaction/client.py

@attr.s  # type:ignore
class AsyncBaseTransactionsClient(abc.ABC):
    """Defines a pattern for implementing the STAC transaction extension."""

    @abc.abstractmethod
    async def create_item(
        self,
        collection_id: str,
        item: Union[Item, ItemCollection],
        **kwargs,
    ) -> Optional[Union[stac.Item, Response, None]]:
        """Create a new item.

        Called with `POST /collections/{collection_id}/items`.

        Args:
            item: the item or item collection
            collection_id: the id of the collection from the resource path

        Returns:
            The item that was created or None if item collection.
        """
        ...

    @abc.abstractmethod
    async def update_item(
        self, collection_id: str, item_id: str, item: Item, **kwargs
    ) -> Optional[Union[stac.Item, Response]]:
        """Perform a complete update on an existing item.

        Called with `PUT /collections/{collection_id}/items`. It is expected
        that this item already exists.  The update should do a diff against the
        saved item and perform any necessary updates.  Partial updates are not
        supported by the transactions extension.

        Args:
            item: the item (must be complete)

        Returns:
            The updated item.
        """
        ...

    @abc.abstractmethod
    async def patch_item(
        self,
        collection_id: str,
        item_id: str,
        patch: Union[PartialItem, List[PatchOperation]],
        **kwargs,
    ) -> Optional[Union[stac.Item, Response]]:
        """Update an item from a collection.

        Called with `PATCH /collections/{collection_id}/items/{item_id}`

        example:
            # convert merge patch item to list of operations
            if isinstance(patch, PartialItem):
                patch = patch.operations()

            item = backend.patch_item(collection_id, item_id, patch)

            return item

        Args:
            item_id: id of the item.
            collection_id: id of the collection.
            patch: either the partial item or list of patch operations.

        Returns:
            The patched item.
        """
        ...

    @abc.abstractmethod
    async def delete_item(
        self, item_id: str, collection_id: str, **kwargs
    ) -> Optional[Union[stac.Item, Response]]:
        """Delete an item from a collection.

        Called with `DELETE /collections/{collection_id}/items/{item_id}`

        Args:
            item_id: id of the item.
            collection_id: id of the collection.

        Returns:
            The deleted item.
        """
        ...

    @abc.abstractmethod
    async def create_collection(
        self, collection: Collection, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Create a new collection.

        Called with `POST /collections`.

        Args:
            collection: the collection

        Returns:
            The collection that was created.
        """
        ...

    @abc.abstractmethod
    async def update_collection(
        self, collection_id: str, collection: Collection, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Perform a complete update on an existing collection.

        Called with `PUT /collections/{collection_id}`. It is expected that this item
        already exists.  The update should do a diff against the saved collection and
        perform any necessary updates.  Partial updates are not supported by the
        transactions extension.

        Args:
            collection_id: id of the existing collection to be updated
            collection: the updated collection (must be complete)

        Returns:
            The updated collection.
        """
        ...

    @abc.abstractmethod
    async def patch_collection(
        self,
        collection_id: str,
        patch: Union[PartialCollection, List[PatchOperation]],
        **kwargs,
    ) -> Optional[Union[stac.Collection, Response]]:
        """Update a collection.

        Called with `PATCH /collections/{collection_id}`

        example:
            # convert merge patch item to list of operations
            if isinstance(patch, PartialCollection):
                patch = patch.operations()

            collection = backend.patch_collection(collection_id, patch)

            return collection

        Args:
            collection_id: id of the collection.
            patch: either the partial collection or list of patch operations.

        Returns:
            The patched collection.
        """
        ...

    @abc.abstractmethod
    async def delete_collection(
        self, collection_id: str, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Delete a collection.

        Called with `DELETE /collections/{collection_id}`

        Args:
            collection_id: id of the collection.

        Returns:
            The deleted collection.
        """
        ...

create_collection abstractmethod async

create_collection(collection: Collection, **kwargs) -> Optional[Union[Collection, Response]]

Create a new collection.

Called with POST /collections.

Parameters:

• collection (Collection) –

  the collection

Returns:

• Optional[Union[Collection, Response]] –

  The collection that was created.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def create_collection(
    self, collection: Collection, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Create a new collection.

    Called with `POST /collections`.

    Args:
        collection: the collection

    Returns:
        The collection that was created.
    """
    ...

create_item abstractmethod async

create_item(
    collection_id: str, item: Union[Item, ItemCollection], **kwargs
) -> Optional[Union[Item, Response, None]]

Create a new item.

Called with POST /collections/{collection_id}/items.

Parameters:

• item (Union[Item, ItemCollection]) –

  the item or item collection

• collection_id (str) –

  the id of the collection from the resource path

Returns:

• Optional[Union[Item, Response, None]] –

  The item that was created or None if item collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def create_item(
    self,
    collection_id: str,
    item: Union[Item, ItemCollection],
    **kwargs,
) -> Optional[Union[stac.Item, Response, None]]:
    """Create a new item.

    Called with `POST /collections/{collection_id}/items`.

    Args:
        item: the item or item collection
        collection_id: the id of the collection from the resource path

    Returns:
        The item that was created or None if item collection.
    """
    ...

delete_collection abstractmethod async

delete_collection(collection_id: str, **kwargs) -> Optional[Union[Collection, Response]]

Delete a collection.

Called with DELETE /collections/{collection_id}

Parameters:

• collection_id (str) –

  id of the collection.

Returns:

• Optional[Union[Collection, Response]] –

  The deleted collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def delete_collection(
    self, collection_id: str, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Delete a collection.

    Called with `DELETE /collections/{collection_id}`

    Args:
        collection_id: id of the collection.

    Returns:
        The deleted collection.
    """
    ...

delete_item abstractmethod async

delete_item(item_id: str, collection_id: str, **kwargs) -> Optional[Union[Item, Response]]

Delete an item from a collection.

Called with DELETE /collections/{collection_id}/items/{item_id}

Parameters:

• item_id (str) –

  id of the item.

• collection_id (str) –

  id of the collection.

Returns:

• Optional[Union[Item, Response]] –

  The deleted item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def delete_item(
    self, item_id: str, collection_id: str, **kwargs
) -> Optional[Union[stac.Item, Response]]:
    """Delete an item from a collection.

    Called with `DELETE /collections/{collection_id}/items/{item_id}`

    Args:
        item_id: id of the item.
        collection_id: id of the collection.

    Returns:
        The deleted item.
    """
    ...

patch_collection abstractmethod async

patch_collection(
    collection_id: str, patch: Union[PartialCollection, List[PatchOperation]], **kwargs
) -> Optional[Union[Collection, Response]]

Update a collection.

Called with PATCH /collections/{collection_id}

example

convert merge patch item to list of operations

if isinstance(patch, PartialCollection): patch = patch.operations()

collection = backend.patch_collection(collection_id, patch)

return collection

Parameters:

• collection_id (str) –

  id of the collection.

• patch (Union[PartialCollection, List[PatchOperation]]) –

  either the partial collection or list of patch operations.

Returns:

• Optional[Union[Collection, Response]] –

  The patched collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def patch_collection(
    self,
    collection_id: str,
    patch: Union[PartialCollection, List[PatchOperation]],
    **kwargs,
) -> Optional[Union[stac.Collection, Response]]:
    """Update a collection.

    Called with `PATCH /collections/{collection_id}`

    example:
        # convert merge patch item to list of operations
        if isinstance(patch, PartialCollection):
            patch = patch.operations()

        collection = backend.patch_collection(collection_id, patch)

        return collection

    Args:
        collection_id: id of the collection.
        patch: either the partial collection or list of patch operations.

    Returns:
        The patched collection.
    """
    ...

patch_item abstractmethod async

patch_item(
    collection_id: str, item_id: str, patch: Union[PartialItem, List[PatchOperation]], **kwargs
) -> Optional[Union[Item, Response]]

Update an item from a collection.

Called with PATCH /collections/{collection_id}/items/{item_id}

example

convert merge patch item to list of operations

if isinstance(patch, PartialItem): patch = patch.operations()

item = backend.patch_item(collection_id, item_id, patch)

return item

Parameters:

• item_id (str) –

  id of the item.

• collection_id (str) –

  id of the collection.

• patch (Union[PartialItem, List[PatchOperation]]) –

  either the partial item or list of patch operations.

Returns:

• Optional[Union[Item, Response]] –

  The patched item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def patch_item(
    self,
    collection_id: str,
    item_id: str,
    patch: Union[PartialItem, List[PatchOperation]],
    **kwargs,
) -> Optional[Union[stac.Item, Response]]:
    """Update an item from a collection.

    Called with `PATCH /collections/{collection_id}/items/{item_id}`

    example:
        # convert merge patch item to list of operations
        if isinstance(patch, PartialItem):
            patch = patch.operations()

        item = backend.patch_item(collection_id, item_id, patch)

        return item

    Args:
        item_id: id of the item.
        collection_id: id of the collection.
        patch: either the partial item or list of patch operations.

    Returns:
        The patched item.
    """
    ...

update_collection abstractmethod async

update_collection(
    collection_id: str, collection: Collection, **kwargs
) -> Optional[Union[Collection, Response]]

Perform a complete update on an existing collection.

Called with PUT /collections/{collection_id}. It is expected that this item already exists. The update should do a diff against the saved collection and perform any necessary updates. Partial updates are not supported by the transactions extension.

Parameters:

• collection_id (str) –

  id of the existing collection to be updated

• collection (Collection) –

  the updated collection (must be complete)

Returns:

• Optional[Union[Collection, Response]] –

  The updated collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def update_collection(
    self, collection_id: str, collection: Collection, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Perform a complete update on an existing collection.

    Called with `PUT /collections/{collection_id}`. It is expected that this item
    already exists.  The update should do a diff against the saved collection and
    perform any necessary updates.  Partial updates are not supported by the
    transactions extension.

    Args:
        collection_id: id of the existing collection to be updated
        collection: the updated collection (must be complete)

    Returns:
        The updated collection.
    """
    ...

update_item abstractmethod async

update_item(
    collection_id: str, item_id: str, item: Item, **kwargs
) -> Optional[Union[Item, Response]]

Perform a complete update on an existing item.

Called with PUT /collections/{collection_id}/items. It is expected that this item already exists. The update should do a diff against the saved item and perform any necessary updates. Partial updates are not supported by the transactions extension.

Parameters:

• item (Item) –

  the item (must be complete)

Returns:

• Optional[Union[Item, Response]] –

  The updated item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
async def update_item(
    self, collection_id: str, item_id: str, item: Item, **kwargs
) -> Optional[Union[stac.Item, Response]]:
    """Perform a complete update on an existing item.

    Called with `PUT /collections/{collection_id}/items`. It is expected
    that this item already exists.  The update should do a diff against the
    saved item and perform any necessary updates.  Partial updates are not
    supported by the transactions extension.

    Args:
        item: the item (must be complete)

    Returns:
        The updated item.
    """
    ...

BaseTransactionsClient

Bases: ABC

Defines a pattern for implementing the STAC API Transaction Extension.

Source code in stac_fastapi/extensions/core/transaction/client.py

@attr.s  # type:ignore
class BaseTransactionsClient(abc.ABC):
    """Defines a pattern for implementing the STAC API Transaction Extension."""

    @abc.abstractmethod
    def create_item(
        self,
        collection_id: str,
        item: Union[Item, ItemCollection],
        **kwargs,
    ) -> Optional[Union[stac.Item, Response, None]]:
        """Create a new item.

        Called with `POST /collections/{collection_id}/items`.

        Args:
            item: the item or item collection
            collection_id: the id of the collection from the resource path

        Returns:
            The item that was created or None if item collection.
        """
        ...

    @abc.abstractmethod
    def update_item(
        self, collection_id: str, item_id: str, item: Item, **kwargs
    ) -> Optional[Union[stac.Item, Response]]:
        """Perform a complete update on an existing item.

        Called with `PUT /collections/{collection_id}/items`. It is expected
        that this item already exists.  The update should do a diff against the
        saved item and perform any necessary updates.  Partial updates are not
        supported by the transactions extension.

        Args:
            item: the item (must be complete)
            collection_id: the id of the collection from the resource path

        Returns:
            The updated item.
        """
        ...

    @abc.abstractmethod
    def patch_item(
        self,
        collection_id: str,
        item_id: str,
        patch: Union[PartialItem, List[PatchOperation]],
        **kwargs,
    ) -> Optional[Union[stac.Item, Response]]:
        """Update an item from a collection.

        Called with `PATCH /collections/{collection_id}/items/{item_id}`

        example:
            # convert merge patch item to list of operations
            if isinstance(patch, PartialItem):
                patch = patch.operations()

            item = backend.patch_item(collection_id, item_id, patch)

            return item

        Args:
            item_id: id of the item.
            collection_id: id of the collection.
            patch: either the partial item or list of patch operations.

        Returns:
            The patched item.
        """
        ...

    @abc.abstractmethod
    def delete_item(
        self, item_id: str, collection_id: str, **kwargs
    ) -> Optional[Union[stac.Item, Response]]:
        """Delete an item from a collection.

        Called with `DELETE /collections/{collection_id}/items/{item_id}`

        Args:
            item_id: id of the item.
            collection_id: id of the collection.

        Returns:
            The deleted item.
        """
        ...

    @abc.abstractmethod
    def create_collection(
        self, collection: Collection, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Create a new collection.

        Called with `POST /collections`.

        Args:
            collection: the collection

        Returns:
            The collection that was created.
        """
        ...

    @abc.abstractmethod
    def update_collection(
        self, collection_id: str, collection: Collection, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Perform a complete update on an existing collection.

        Called with `PUT /collections/{collection_id}`. It is expected that this
        collection already exists.  The update should do a diff against the saved
        collection and perform any necessary updates.  Partial updates are not
        supported by the transactions extension.

        Args:
            collection_id: id of the existing collection to be updated
            collection: the updated collection (must be complete)

        Returns:
            The updated collection.
        """
        ...

    @abc.abstractmethod
    def patch_collection(
        self,
        collection_id: str,
        patch: Union[PartialCollection, List[PatchOperation]],
        **kwargs,
    ) -> Optional[Union[stac.Collection, Response]]:
        """Update a collection.

        Called with `PATCH /collections/{collection_id}`

        example:
            # convert merge patch item to list of operations
            if isinstance(patch, PartialCollection):
                patch = patch.operations()

            collection = backend.patch_collection(collection_id, patch)

            return collection

        Args:
            collection_id: id of the collection.
            patch: either the partial collection or list of patch operations.

        Returns:
            The patched collection.
        """
        ...

    @abc.abstractmethod
    def delete_collection(
        self, collection_id: str, **kwargs
    ) -> Optional[Union[stac.Collection, Response]]:
        """Delete a collection.

        Called with `DELETE /collections/{collection_id}`

        Args:
            collection_id: id of the collection.

        Returns:
            The deleted collection.
        """
        ...

create_collection abstractmethod

create_collection(collection: Collection, **kwargs) -> Optional[Union[Collection, Response]]

Create a new collection.

Called with POST /collections.

Parameters:

• collection (Collection) –

  the collection

Returns:

• Optional[Union[Collection, Response]] –

  The collection that was created.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def create_collection(
    self, collection: Collection, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Create a new collection.

    Called with `POST /collections`.

    Args:
        collection: the collection

    Returns:
        The collection that was created.
    """
    ...

create_item abstractmethod

create_item(
    collection_id: str, item: Union[Item, ItemCollection], **kwargs
) -> Optional[Union[Item, Response, None]]

Create a new item.

Called with POST /collections/{collection_id}/items.

Parameters:

• item (Union[Item, ItemCollection]) –

  the item or item collection

• collection_id (str) –

  the id of the collection from the resource path

Returns:

• Optional[Union[Item, Response, None]] –

  The item that was created or None if item collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def create_item(
    self,
    collection_id: str,
    item: Union[Item, ItemCollection],
    **kwargs,
) -> Optional[Union[stac.Item, Response, None]]:
    """Create a new item.

    Called with `POST /collections/{collection_id}/items`.

    Args:
        item: the item or item collection
        collection_id: the id of the collection from the resource path

    Returns:
        The item that was created or None if item collection.
    """
    ...

delete_collection abstractmethod

delete_collection(collection_id: str, **kwargs) -> Optional[Union[Collection, Response]]

Delete a collection.

Called with DELETE /collections/{collection_id}

Parameters:

• collection_id (str) –

  id of the collection.

Returns:

• Optional[Union[Collection, Response]] –

  The deleted collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def delete_collection(
    self, collection_id: str, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Delete a collection.

    Called with `DELETE /collections/{collection_id}`

    Args:
        collection_id: id of the collection.

    Returns:
        The deleted collection.
    """
    ...

delete_item abstractmethod

delete_item(item_id: str, collection_id: str, **kwargs) -> Optional[Union[Item, Response]]

Delete an item from a collection.

Called with DELETE /collections/{collection_id}/items/{item_id}

Parameters:

• item_id (str) –

  id of the item.

• collection_id (str) –

  id of the collection.

Returns:

• Optional[Union[Item, Response]] –

  The deleted item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def delete_item(
    self, item_id: str, collection_id: str, **kwargs
) -> Optional[Union[stac.Item, Response]]:
    """Delete an item from a collection.

    Called with `DELETE /collections/{collection_id}/items/{item_id}`

    Args:
        item_id: id of the item.
        collection_id: id of the collection.

    Returns:
        The deleted item.
    """
    ...

patch_collection abstractmethod

patch_collection(
    collection_id: str, patch: Union[PartialCollection, List[PatchOperation]], **kwargs
) -> Optional[Union[Collection, Response]]

Update a collection.

Called with PATCH /collections/{collection_id}

example

convert merge patch item to list of operations

if isinstance(patch, PartialCollection): patch = patch.operations()

collection = backend.patch_collection(collection_id, patch)

return collection

Parameters:

• collection_id (str) –

  id of the collection.

• patch (Union[PartialCollection, List[PatchOperation]]) –

  either the partial collection or list of patch operations.

Returns:

• Optional[Union[Collection, Response]] –

  The patched collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def patch_collection(
    self,
    collection_id: str,
    patch: Union[PartialCollection, List[PatchOperation]],
    **kwargs,
) -> Optional[Union[stac.Collection, Response]]:
    """Update a collection.

    Called with `PATCH /collections/{collection_id}`

    example:
        # convert merge patch item to list of operations
        if isinstance(patch, PartialCollection):
            patch = patch.operations()

        collection = backend.patch_collection(collection_id, patch)

        return collection

    Args:
        collection_id: id of the collection.
        patch: either the partial collection or list of patch operations.

    Returns:
        The patched collection.
    """
    ...

patch_item abstractmethod

patch_item(
    collection_id: str, item_id: str, patch: Union[PartialItem, List[PatchOperation]], **kwargs
) -> Optional[Union[Item, Response]]

Update an item from a collection.

Called with PATCH /collections/{collection_id}/items/{item_id}

example

convert merge patch item to list of operations

if isinstance(patch, PartialItem): patch = patch.operations()

item = backend.patch_item(collection_id, item_id, patch)

return item

Parameters:

• item_id (str) –

  id of the item.

• collection_id (str) –

  id of the collection.

• patch (Union[PartialItem, List[PatchOperation]]) –

  either the partial item or list of patch operations.

Returns:

• Optional[Union[Item, Response]] –

  The patched item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def patch_item(
    self,
    collection_id: str,
    item_id: str,
    patch: Union[PartialItem, List[PatchOperation]],
    **kwargs,
) -> Optional[Union[stac.Item, Response]]:
    """Update an item from a collection.

    Called with `PATCH /collections/{collection_id}/items/{item_id}`

    example:
        # convert merge patch item to list of operations
        if isinstance(patch, PartialItem):
            patch = patch.operations()

        item = backend.patch_item(collection_id, item_id, patch)

        return item

    Args:
        item_id: id of the item.
        collection_id: id of the collection.
        patch: either the partial item or list of patch operations.

    Returns:
        The patched item.
    """
    ...

update_collection abstractmethod

update_collection(
    collection_id: str, collection: Collection, **kwargs
) -> Optional[Union[Collection, Response]]

Perform a complete update on an existing collection.

Called with PUT /collections/{collection_id}. It is expected that this collection already exists. The update should do a diff against the saved collection and perform any necessary updates. Partial updates are not supported by the transactions extension.

Parameters:

• collection_id (str) –

  id of the existing collection to be updated

• collection (Collection) –

  the updated collection (must be complete)

Returns:

• Optional[Union[Collection, Response]] –

  The updated collection.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def update_collection(
    self, collection_id: str, collection: Collection, **kwargs
) -> Optional[Union[stac.Collection, Response]]:
    """Perform a complete update on an existing collection.

    Called with `PUT /collections/{collection_id}`. It is expected that this
    collection already exists.  The update should do a diff against the saved
    collection and perform any necessary updates.  Partial updates are not
    supported by the transactions extension.

    Args:
        collection_id: id of the existing collection to be updated
        collection: the updated collection (must be complete)

    Returns:
        The updated collection.
    """
    ...

update_item abstractmethod

update_item(
    collection_id: str, item_id: str, item: Item, **kwargs
) -> Optional[Union[Item, Response]]

Perform a complete update on an existing item.

Called with PUT /collections/{collection_id}/items. It is expected that this item already exists. The update should do a diff against the saved item and perform any necessary updates. Partial updates are not supported by the transactions extension.

Parameters:

• item (Item) –

  the item (must be complete)

• collection_id (str) –

  the id of the collection from the resource path

Returns:

• Optional[Union[Item, Response]] –

  The updated item.

Source code in stac_fastapi/extensions/core/transaction/client.py

@abc.abstractmethod
def update_item(
    self, collection_id: str, item_id: str, item: Item, **kwargs
) -> Optional[Union[stac.Item, Response]]:
    """Perform a complete update on an existing item.

    Called with `PUT /collections/{collection_id}/items`. It is expected
    that this item already exists.  The update should do a diff against the
    saved item and perform any necessary updates.  Partial updates are not
    supported by the transactions extension.

    Args:
        item: the item (must be complete)
        collection_id: the id of the collection from the resource path

    Returns:
        The updated item.
    """
    ...

TransactionConformanceClasses

Bases: str, Enum

Conformance classes for the Transaction extension.

See stac-api-extensions/transaction

Source code in stac_fastapi/extensions/core/transaction/transaction.py

class TransactionConformanceClasses(str, Enum):
    """Conformance classes for the Transaction extension.

    See https://github.com/stac-api-extensions/transaction

    """

    ITEMS = "https://api.stacspec.org/v1.0.0/ogcapi-features/extensions/transaction"
    COLLECTIONS = "https://api.stacspec.org/v1.0.0/collections/extensions/transaction"

TransactionExtension

Bases: ApiExtension

Transaction Extension.

The transaction extension adds several endpoints which allow the creation, deletion, and updating of items and collections: POST /collections PUT /collections/{collection_id} DELETE /collections/{collection_id} POST /collections/{collection_id}/items PUT /collections/{collection_id}/items DELETE /collections/{collection_id}/items

stac-api-extensions/transaction stac-api-extensions/collection-transaction

Attributes:

• client (Union[AsyncBaseTransactionsClient, BaseTransactionsClient]) –

  CRUD application logic

Source code in stac_fastapi/extensions/core/transaction/transaction.py

@attr.s
class TransactionExtension(ApiExtension):
    """Transaction Extension.

    The transaction extension adds several endpoints which allow the creation,
    deletion, and updating of items and collections:
        POST /collections
        PUT /collections/{collection_id}
        DELETE /collections/{collection_id}
        POST /collections/{collection_id}/items
        PUT /collections/{collection_id}/items
        DELETE /collections/{collection_id}/items

    https://github.com/stac-api-extensions/transaction
    https://github.com/stac-api-extensions/collection-transaction

    Attributes:
        client: CRUD application logic

    """

    client: Union[AsyncBaseTransactionsClient, BaseTransactionsClient] = attr.ib()
    settings: ApiSettings = attr.ib()
    conformance_classes: List[str] = attr.ib(
        factory=lambda: [
            TransactionConformanceClasses.ITEMS,
            TransactionConformanceClasses.COLLECTIONS,
        ]
    )
    schema_href: Optional[str] = attr.ib(default=None)
    router: APIRouter = attr.ib(factory=APIRouter)
    response_class: Type[Response] = attr.ib(default=JSONResponse)

    def register_create_item(self):
        """Register create item endpoint (POST /collections/{collection_id}/items)."""
        self.router.add_api_route(
            name="Create Item",
            path="/collections/{collection_id}/items",
            status_code=201,
            response_model=Item if self.settings.enable_response_models else None,
            responses={
                201: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": Item,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["POST"],
            endpoint=create_async_endpoint(self.client.create_item, PostItem),
        )

    def register_update_item(self):
        """Register update item endpoint (PUT
        /collections/{collection_id}/items/{item_id})."""
        self.router.add_api_route(
            name="Update Item",
            path="/collections/{collection_id}/items/{item_id}",
            response_model=Item if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": Item,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["PUT"],
            endpoint=create_async_endpoint(self.client.update_item, PutItem),
        )

    def register_patch_item(self):
        """Register patch item endpoint (PATCH
        /collections/{collection_id}/items/{item_id})."""
        self.router.add_api_route(
            name="Patch Item",
            path="/collections/{collection_id}/items/{item_id}",
            response_model=Item if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": Item,
                }
            },
            openapi_extra={
                "requestBody": {
                    "content": {
                        "application/json-patch+json": {
                            "schema": _patch_item_schema,
                        },
                        "application/merge-patch+json": {
                            "schema": PartialItem.model_json_schema(),
                        },
                        "application/json": {
                            "schema": PartialItem.model_json_schema(),
                        },
                    },
                    "required": True,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["PATCH"],
            endpoint=create_async_endpoint(
                self.client.patch_item,
                PatchItem,
            ),
        )

    def register_delete_item(self):
        """Register delete item endpoint (DELETE
        /collections/{collection_id}/items/{item_id})."""
        self.router.add_api_route(
            name="Delete Item",
            path="/collections/{collection_id}/items/{item_id}",
            response_model=Item if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": Item,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["DELETE"],
            endpoint=create_async_endpoint(self.client.delete_item, ItemUri),
        )

    def register_create_collection(self):
        """Register create collection endpoint (POST /collections)."""
        self.router.add_api_route(
            name="Create Collection",
            path="/collections",
            status_code=201,
            response_model=Collection if self.settings.enable_response_models else None,
            responses={
                201: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": Collection,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["POST"],
            endpoint=create_async_endpoint(self.client.create_collection, Collection),
        )

    def register_update_collection(self):
        """Register update collection endpoint (PUT /collections/{collection_id})."""
        self.router.add_api_route(
            name="Update Collection",
            path="/collections/{collection_id}",
            response_model=Collection if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": Collection,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["PUT"],
            endpoint=create_async_endpoint(self.client.update_collection, PutCollection),
        )

    def register_patch_collection(self):
        """Register patch collection endpoint (PATCH /collections/{collection_id})."""
        self.router.add_api_route(
            name="Patch Collection",
            path="/collections/{collection_id}",
            response_model=Collection if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.geojson.value: {},
                    },
                    "model": Collection,
                }
            },
            openapi_extra={
                "requestBody": {
                    "content": {
                        "application/json-patch+json": {
                            "schema": _patch_collection_schema,
                        },
                        "application/merge-patch+json": {
                            "schema": PartialCollection.model_json_schema(),
                        },
                        "application/json": {
                            "schema": PartialCollection.model_json_schema(),
                        },
                    },
                    "required": True,
                },
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["PATCH"],
            endpoint=create_async_endpoint(
                self.client.patch_collection,
                PatchCollection,
            ),
        )

    def register_delete_collection(self):
        """Register delete collection endpoint (DELETE /collections/{collection_id})."""
        self.router.add_api_route(
            name="Delete Collection",
            path="/collections/{collection_id}",
            response_model=Collection if self.settings.enable_response_models else None,
            responses={
                200: {
                    "content": {
                        MimeTypes.json.value: {},
                    },
                    "model": Collection,
                }
            },
            response_class=self.response_class,
            response_model_exclude_unset=True,
            response_model_exclude_none=True,
            methods=["DELETE"],
            endpoint=create_async_endpoint(self.client.delete_collection, CollectionUri),
        )

    def register(self, app: FastAPI) -> None:
        """Register the extension with a FastAPI application.

        Args:
            app: target FastAPI application.

        Returns:
            None
        """
        self.router.prefix = app.state.router_prefix
        self.register_create_item()
        self.register_update_item()
        self.register_patch_item()
        self.register_delete_item()
        self.register_create_collection()
        self.register_update_collection()
        self.register_patch_collection()
        self.register_delete_collection()
        app.include_router(self.router, tags=["Transaction Extension"])

register

register(app: FastAPI) -> None

Register the extension with a FastAPI application.

Parameters:

• app (FastAPI) –

  target FastAPI application.

Returns:

• None –

  None

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register(self, app: FastAPI) -> None:
    """Register the extension with a FastAPI application.

    Args:
        app: target FastAPI application.

    Returns:
        None
    """
    self.router.prefix = app.state.router_prefix
    self.register_create_item()
    self.register_update_item()
    self.register_patch_item()
    self.register_delete_item()
    self.register_create_collection()
    self.register_update_collection()
    self.register_patch_collection()
    self.register_delete_collection()
    app.include_router(self.router, tags=["Transaction Extension"])

register_create_collection

register_create_collection()

Register create collection endpoint (POST /collections).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_create_collection(self):
    """Register create collection endpoint (POST /collections)."""
    self.router.add_api_route(
        name="Create Collection",
        path="/collections",
        status_code=201,
        response_model=Collection if self.settings.enable_response_models else None,
        responses={
            201: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": Collection,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["POST"],
        endpoint=create_async_endpoint(self.client.create_collection, Collection),
    )

register_create_item

register_create_item()

Register create item endpoint (POST /collections/{collection_id}/items).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_create_item(self):
    """Register create item endpoint (POST /collections/{collection_id}/items)."""
    self.router.add_api_route(
        name="Create Item",
        path="/collections/{collection_id}/items",
        status_code=201,
        response_model=Item if self.settings.enable_response_models else None,
        responses={
            201: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": Item,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["POST"],
        endpoint=create_async_endpoint(self.client.create_item, PostItem),
    )

register_delete_collection

register_delete_collection()

Register delete collection endpoint (DELETE /collections/{collection_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_delete_collection(self):
    """Register delete collection endpoint (DELETE /collections/{collection_id})."""
    self.router.add_api_route(
        name="Delete Collection",
        path="/collections/{collection_id}",
        response_model=Collection if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": Collection,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["DELETE"],
        endpoint=create_async_endpoint(self.client.delete_collection, CollectionUri),
    )

register_delete_item

register_delete_item()

Register delete item endpoint (DELETE /collections/{collection_id}/items/{item_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_delete_item(self):
    """Register delete item endpoint (DELETE
    /collections/{collection_id}/items/{item_id})."""
    self.router.add_api_route(
        name="Delete Item",
        path="/collections/{collection_id}/items/{item_id}",
        response_model=Item if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": Item,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["DELETE"],
        endpoint=create_async_endpoint(self.client.delete_item, ItemUri),
    )

register_patch_collection

register_patch_collection()

Register patch collection endpoint (PATCH /collections/{collection_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_patch_collection(self):
    """Register patch collection endpoint (PATCH /collections/{collection_id})."""
    self.router.add_api_route(
        name="Patch Collection",
        path="/collections/{collection_id}",
        response_model=Collection if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": Collection,
            }
        },
        openapi_extra={
            "requestBody": {
                "content": {
                    "application/json-patch+json": {
                        "schema": _patch_collection_schema,
                    },
                    "application/merge-patch+json": {
                        "schema": PartialCollection.model_json_schema(),
                    },
                    "application/json": {
                        "schema": PartialCollection.model_json_schema(),
                    },
                },
                "required": True,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["PATCH"],
        endpoint=create_async_endpoint(
            self.client.patch_collection,
            PatchCollection,
        ),
    )

register_patch_item

register_patch_item()

Register patch item endpoint (PATCH /collections/{collection_id}/items/{item_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_patch_item(self):
    """Register patch item endpoint (PATCH
    /collections/{collection_id}/items/{item_id})."""
    self.router.add_api_route(
        name="Patch Item",
        path="/collections/{collection_id}/items/{item_id}",
        response_model=Item if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": Item,
            }
        },
        openapi_extra={
            "requestBody": {
                "content": {
                    "application/json-patch+json": {
                        "schema": _patch_item_schema,
                    },
                    "application/merge-patch+json": {
                        "schema": PartialItem.model_json_schema(),
                    },
                    "application/json": {
                        "schema": PartialItem.model_json_schema(),
                    },
                },
                "required": True,
            },
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["PATCH"],
        endpoint=create_async_endpoint(
            self.client.patch_item,
            PatchItem,
        ),
    )

register_update_collection

register_update_collection()

Register update collection endpoint (PUT /collections/{collection_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_update_collection(self):
    """Register update collection endpoint (PUT /collections/{collection_id})."""
    self.router.add_api_route(
        name="Update Collection",
        path="/collections/{collection_id}",
        response_model=Collection if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.json.value: {},
                },
                "model": Collection,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["PUT"],
        endpoint=create_async_endpoint(self.client.update_collection, PutCollection),
    )

register_update_item

register_update_item()

Register update item endpoint (PUT /collections/{collection_id}/items/{item_id}).

Source code in stac_fastapi/extensions/core/transaction/transaction.py

def register_update_item(self):
    """Register update item endpoint (PUT
    /collections/{collection_id}/items/{item_id})."""
    self.router.add_api_route(
        name="Update Item",
        path="/collections/{collection_id}/items/{item_id}",
        response_model=Item if self.settings.enable_response_models else None,
        responses={
            200: {
                "content": {
                    MimeTypes.geojson.value: {},
                },
                "model": Item,
            }
        },
        response_class=self.response_class,
        response_model_exclude_unset=True,
        response_model_exclude_none=True,
        methods=["PUT"],
        endpoint=create_async_endpoint(self.client.update_item, PutItem),
    )