diff --git a/CHANGES.md b/CHANGES.md index 98aa9574..44ba6986 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -4,8 +4,9 @@ ### Changed -* Add Item and Collection `PATCH` endpoints with support for [RFC 6902](https://tools.ietf.org/html/rfc6902) and [RFC 7396](https://tools.ietf.org/html/rfc7386) in the `TransactionExtension` +- Add Item and Collection `PATCH` endpoints with support for [RFC 6902](https://tools.ietf.org/html/rfc6902) and [RFC 7396](https://tools.ietf.org/html/rfc7386) in the `TransactionExtension` - remove support of `cql-json` in Filter extension ([#840](https://github.com/stac-utils/stac-fastapi/pull/840)) +- move `transaction` clients and models to `stac_fastapi.extension` sub-module ### Fixed diff --git a/stac_fastapi/api/tests/test_api.py b/stac_fastapi/api/tests/test_api.py index d4ebf951..5f9aa73e 100644 --- a/stac_fastapi/api/tests/test_api.py +++ b/stac_fastapi/api/tests/test_api.py @@ -4,6 +4,7 @@ from stac_fastapi.api.app import StacApi from stac_fastapi.api.models import ItemCollectionUri, create_request_model from stac_fastapi.extensions.core import TokenPaginationExtension, TransactionExtension +from stac_fastapi.extensions.core.transaction import BaseTransactionsClient from stac_fastapi.types import config, core @@ -418,7 +419,7 @@ def item_collection(self, *args, **kwargs): ... -class DummyTransactionsClient(core.BaseTransactionsClient): +class DummyTransactionsClient(BaseTransactionsClient): """Defines a pattern for implementing the STAC transaction extension.""" def create_item(self, *args, **kwargs): diff --git a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/__init__.py b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/__init__.py new file mode 100644 index 00000000..5e71f956 --- /dev/null +++ b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/__init__.py @@ -0,0 +1,11 @@ +"""transaction extension module.""" + +from .client import AsyncBaseTransactionsClient, BaseTransactionsClient +from .transaction import TransactionConformanceClasses, TransactionExtension + +__all__ = [ + "AsyncBaseTransactionsClient", + "BaseTransactionsClient", + "TransactionExtension", + "TransactionConformanceClasses", +] diff --git a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/client.py b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/client.py new file mode 100644 index 00000000..305e8128 --- /dev/null +++ b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/client.py @@ -0,0 +1,359 @@ +"""Transaction clients.""" + +import abc +from typing import List, Optional, Union + +import attr +from stac_pydantic import Collection, Item, ItemCollection +from starlette.responses import Response + +from stac_fastapi.types import stac + +from .request import PartialCollection, PartialItem, PatchOperation + + +@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. + """ + ... + + +@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. + """ + ... diff --git a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/request.py b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/request.py new file mode 100644 index 00000000..da3b0745 --- /dev/null +++ b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/request.py @@ -0,0 +1,156 @@ +"""Transaction extension request models.""" + +import json +from typing import Any, Dict, List, Literal, Optional, Union + +from pydantic import ConfigDict, Field +from stac_pydantic.shared import BBox, StacBaseModel + + +class PatchAddReplaceTest(StacBaseModel): + """Add, Replace or Test Operation.""" + + model_config = ConfigDict( + json_schema_extra={ + "examples": [ + {"op": "add", "path": "/properties/foo", "value": "bar"}, + {"op": "replace", "path": "/properties/foo", "value": "bar"}, + {"op": "test", "path": "/properties/foo", "value": "bar"}, + ] + } + ) + + path: str + op: Literal["add", "replace", "test"] + value: Any + + @property + def json_value(self) -> str: + """JSON dump of value field. + + Returns: + str: JSON-ised value + """ + return json.dumps(self.value) + + +class PatchRemove(StacBaseModel): + """Remove Operation.""" + + model_config = ConfigDict( + json_schema_extra={ + "examples": [ + { + "op": "remove", + "path": "/properties/foo", + } + ] + } + ) + + path: str + op: Literal["remove"] + + +class PatchMoveCopy(StacBaseModel): + """Move or Copy Operation.""" + + model_config = ConfigDict( + populate_by_name=True, + json_schema_extra={ + "examples": [ + { + "op": "copy", + "path": "/properties/foo", + "from": "/properties/bar", + }, + { + "op": "move", + "path": "/properties/foo", + "from": "/properties/bar", + }, + ] + }, + ) + + path: str + op: Literal["move", "copy"] + from_: str = Field(alias="from") + + +PatchOperation = Union[PatchAddReplaceTest, PatchMoveCopy, PatchRemove] + + +class BasePartial(StacBaseModel): + """Base Partial Class.""" + + @staticmethod + def merge_to_operations(data: Dict) -> List[PatchOperation]: + """Convert merge operation to list of RF6902 operations. + + Args: + data: dictionary to convert. + + Returns: + List: list of RF6902 operations. + """ + operations = [] + + for key, value in data.copy().items(): + if value is None: + operations.append(PatchRemove(op="remove", path=f"/{key}")) + + elif isinstance(value, dict): + nested_operations = BasePartial.merge_to_operations(value) + + for nested_operation in nested_operations: + nested_operation.path = f"/{key}{nested_operation.path}" + operations.append(nested_operation) + + else: + operations.append( + PatchAddReplaceTest(op="add", path=f"/{key}", value=value) + ) + + return operations + + def operations(self) -> List[PatchOperation]: + """Equivalent RF6902 operations to merge of Partial. + + Returns: + List[PatchOperation]: Equivalent list of RF6902 operations + """ + return self.merge_to_operations(self.model_dump()) + + +class PartialCollection(BasePartial): + """Partial STAC Collection.""" + + type: Optional[str] = None + stac_version: Optional[str] = None + stac_extensions: Optional[List[str]] = None + id: Optional[str] = None + title: Optional[str] = None + description: Optional[str] = None + links: Optional[Dict[str, Any]] = None + keywords: Optional[List[str]] = None + license: Optional[str] = None + providers: Optional[List[Dict[str, Any]]] = None + extent: Optional[Dict[str, Any]] = None + summaries: Optional[Dict[str, Any]] = None + assets: Optional[Dict[str, Any]] = None + + +class PartialItem(BasePartial): + """Partial STAC Item.""" + + type: Optional[Literal["Feature"]] = None + stac_version: Optional[str] = None + stac_extensions: Optional[List[str]] = None + id: Optional[str] = None + geometry: Optional[Dict[str, Any]] = None + bbox: Optional[BBox] = None + properties: Optional[Dict[str, Any]] = None + links: Optional[List[Dict[str, Any]]] = None + assets: Optional[Dict[str, Any]] = None + collection: Optional[str] = None diff --git a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/transaction.py similarity index 98% rename from stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py rename to stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/transaction.py index c0e390ae..ca61c9e6 100644 --- a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py +++ b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction/transaction.py @@ -14,9 +14,10 @@ from stac_fastapi.api.models import CollectionUri, ItemUri, JSONResponse from stac_fastapi.api.routes import create_async_endpoint from stac_fastapi.types.config import ApiSettings -from stac_fastapi.types.core import AsyncBaseTransactionsClient, BaseTransactionsClient from stac_fastapi.types.extension import ApiExtension -from stac_fastapi.types.stac import PartialCollection, PartialItem, PatchOperation + +from .client import AsyncBaseTransactionsClient, BaseTransactionsClient +from .request import PartialCollection, PartialItem, PatchOperation class TransactionConformanceClasses(str, Enum): diff --git a/stac_fastapi/extensions/tests/test_transaction.py b/stac_fastapi/extensions/tests/test_transaction.py index 2f6487b7..02f38606 100644 --- a/stac_fastapi/extensions/tests/test_transaction.py +++ b/stac_fastapi/extensions/tests/test_transaction.py @@ -9,9 +9,14 @@ from stac_fastapi.api.app import StacApi from stac_fastapi.extensions.core import TransactionExtension +from stac_fastapi.extensions.core.transaction import BaseTransactionsClient +from stac_fastapi.extensions.core.transaction.request import ( + PartialCollection, + PartialItem, + PatchOperation, +) from stac_fastapi.types.config import ApiSettings -from stac_fastapi.types.core import BaseCoreClient, BaseTransactionsClient -from stac_fastapi.types.stac import PartialCollection, PartialItem, PatchOperation +from stac_fastapi.types.core import BaseCoreClient class DummyCoreClient(BaseCoreClient): diff --git a/stac_fastapi/types/stac_fastapi/types/core.py b/stac_fastapi/types/stac_fastapi/types/core.py index 7d5e1fd9..85a28763 100644 --- a/stac_fastapi/types/stac_fastapi/types/core.py +++ b/stac_fastapi/types/stac_fastapi/types/core.py @@ -7,11 +7,9 @@ import attr from fastapi import Request from geojson_pydantic.geometries import Geometry -from stac_pydantic import Collection, Item, ItemCollection from stac_pydantic.links import Relations from stac_pydantic.shared import BBox, MimeTypes from stac_pydantic.version import STAC_VERSION -from starlette.responses import Response from . import stac from .config import ApiSettings @@ -23,8 +21,6 @@ __all__ = [ "NumType", "StacType", - "BaseTransactionsClient", - "AsyncBaseTransactionsClient", "LandingPageMixin", "BaseCoreClient", "AsyncBaseCoreClient", @@ -36,353 +32,6 @@ api_settings = ApiSettings() -@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[stac.PartialItem, List[stac.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[stac.PartialCollection, List[stac.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. - """ - ... - - -@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[stac.PartialItem, List[stac.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[stac.PartialCollection, List[stac.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. - """ - ... - - @attr.s class LandingPageMixin(abc.ABC): """Create a STAC landing page (GET /).""" diff --git a/stac_fastapi/types/stac_fastapi/types/stac.py b/stac_fastapi/types/stac_fastapi/types/stac.py index dfd9a62b..7cfdb3d6 100644 --- a/stac_fastapi/types/stac_fastapi/types/stac.py +++ b/stac_fastapi/types/stac_fastapi/types/stac.py @@ -1,10 +1,8 @@ """STAC types.""" -import json -from typing import Any, Dict, List, Literal, Optional, Union +from typing import Any, Dict, List, Literal, Union -from pydantic import ConfigDict, Field -from stac_pydantic.shared import BBox, StacBaseModel +from stac_pydantic.shared import BBox from typing_extensions import NotRequired, TypedDict NumType = Union[float, int] @@ -79,152 +77,3 @@ class Collections(TypedDict): links: List[Dict[str, Any]] numberMatched: NotRequired[int] numberReturned: NotRequired[int] - - -class PatchAddReplaceTest(StacBaseModel): - """Add, Replace or Test Operation.""" - - model_config = ConfigDict( - json_schema_extra={ - "examples": [ - {"op": "add", "path": "/properties/foo", "value": "bar"}, - {"op": "replace", "path": "/properties/foo", "value": "bar"}, - {"op": "test", "path": "/properties/foo", "value": "bar"}, - ] - } - ) - - path: str - op: Literal["add", "replace", "test"] - value: Any - - @property - def json_value(self) -> str: - """JSON dump of value field. - - Returns: - str: JSON-ised value - """ - return json.dumps(self.value) - - -class PatchRemove(StacBaseModel): - """Remove Operation.""" - - model_config = ConfigDict( - json_schema_extra={ - "examples": [ - { - "op": "remove", - "path": "/properties/foo", - } - ] - } - ) - - path: str - op: Literal["remove"] - - -class PatchMoveCopy(StacBaseModel): - """Move or Copy Operation.""" - - model_config = ConfigDict( - populate_by_name=True, - json_schema_extra={ - "examples": [ - { - "op": "copy", - "path": "/properties/foo", - "from": "/properties/bar", - }, - { - "op": "move", - "path": "/properties/foo", - "from": "/properties/bar", - }, - ] - }, - ) - - path: str - op: Literal["move", "copy"] - from_: str = Field(alias="from") - - -PatchOperation = Union[PatchAddReplaceTest, PatchMoveCopy, PatchRemove] - - -class BasePartial(StacBaseModel): - """Base Partial Class.""" - - @staticmethod - def merge_to_operations(data: Dict) -> List[PatchOperation]: - """Convert merge operation to list of RF6902 operations. - - Args: - data: dictionary to convert. - - Returns: - List: list of RF6902 operations. - """ - operations = [] - - for key, value in data.copy().items(): - if value is None: - operations.append(PatchRemove(op="remove", path=f"/{key}")) - - elif isinstance(value, dict): - nested_operations = BasePartial.merge_to_operations(value) - - for nested_operation in nested_operations: - nested_operation.path = f"/{key}{nested_operation.path}" - operations.append(nested_operation) - - else: - operations.append( - PatchAddReplaceTest(op="add", path=f"/{key}", value=value) - ) - - return operations - - def operations(self) -> List[PatchOperation]: - """Equivalent RF6902 operations to merge of Partial. - - Returns: - List[PatchOperation]: Equivalent list of RF6902 operations - """ - return self.merge_to_operations(self.model_dump()) - - -class PartialCollection(BasePartial): - """Partial STAC Collection.""" - - type: Optional[str] = None - stac_version: Optional[str] = None - stac_extensions: Optional[List[str]] = None - id: Optional[str] = None - title: Optional[str] = None - description: Optional[str] = None - links: Optional[Dict[str, Any]] = None - keywords: Optional[List[str]] = None - license: Optional[str] = None - providers: Optional[List[Dict[str, Any]]] = None - extent: Optional[Dict[str, Any]] = None - summaries: Optional[Dict[str, Any]] = None - assets: Optional[Dict[str, Any]] = None - - -class PartialItem(BasePartial): - """Partial STAC Item.""" - - type: Optional[Literal["Feature"]] = None - stac_version: Optional[str] = None - stac_extensions: Optional[List[str]] = None - id: Optional[str] = None - geometry: Optional[Dict[str, Any]] = None - bbox: Optional[BBox] = None - properties: Optional[Dict[str, Any]] = None - links: Optional[List[Dict[str, Any]]] = None - assets: Optional[Dict[str, Any]] = None - collection: Optional[str] = None