Update request body types for Transaction Extension spec and client (#574)

* Update request body for Transaction Extension spec

A POST against an items endpoint should accept an Item or ItemCollection

https://github.com/radiantearth/stac-api-spec/blob/master/ogcapi-features/extensions/transaction/README.md#methods

Add literal values to the stac_types `type` attribute in order to allow fastapi a
discriminator value for parsing the request body to the correct TypedDict.

* Changelog

* feat: add a simple smoke test

It doesn't exercise much, but at least it demonstrates what we're trying to do.

---------

Co-authored-by: Pete Gadomski <[email protected]>

Author: mmcfarland

CHANGES.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Added
+
+* Add support for POSTing ItemCollections to the /items endpoint of the Transaction Extension ([#547](https://github.com/stac-utils/stac-fastapi/pull/574)
+
 ## [2.4.6] - 2023-05-09
 
 ### Changed

stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py

@@ -18,7 +18,9 @@
 class PostItem(CollectionUri):
     """Create Item."""
 
-    item: stac_types.Item = attr.ib(default=Body(None))
+    item: Union[stac_types.Item, stac_types.ItemCollection] = attr.ib(
+        default=Body(None)
+    )
 
 
 @attr.s

stac_fastapi/extensions/tests/test_transaction.py

@@ -0,0 +1,121 @@
+import json
+from typing import Iterator, Union
+
+import pytest
+from starlette.testclient import TestClient
+
+from stac_fastapi.api.app import StacApi
+from stac_fastapi.extensions.core import TransactionExtension
+from stac_fastapi.types.config import ApiSettings
+from stac_fastapi.types.core import BaseCoreClient, BaseTransactionsClient
+from stac_fastapi.types.stac import Item, ItemCollection
+
+
+class DummyCoreClient(BaseCoreClient):
+    def all_collections(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def get_collection(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def get_item(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def get_search(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def post_search(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def item_collection(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class DummyTransactionsClient(BaseTransactionsClient):
+    """Defines a pattern for implementing the STAC transaction extension."""
+
+    def create_item(self, item: Union[Item, ItemCollection], *args, **kwargs):
+        return {"created": True, "type": item["type"]}
+
+    def update_item(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def delete_item(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def create_collection(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def update_collection(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def delete_collection(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+def test_create_item(client: TestClient, item: Item) -> None:
+    response = client.post("/collections/a-collection/items", content=json.dumps(item))
+    assert response.is_success, response.text
+    assert response.json()["type"] == "Feature"
+
+
+def test_create_item_collection(
+    client: TestClient, item_collection: ItemCollection
+) -> None:
+    response = client.post(
+        "/collections/a-collection/items", content=json.dumps(item_collection)
+    )
+    assert response.is_success, response.text
+    assert response.json()["type"] == "FeatureCollection"
+
+
+@pytest.fixture
+def client(
+    core_client: DummyCoreClient, transactions_client: DummyTransactionsClient
+) -> Iterator[TestClient]:
+    settings = ApiSettings()
+    api = StacApi(
+        settings=settings,
+        client=core_client,
+        extensions=[
+            TransactionExtension(client=transactions_client, settings=settings),
+        ],
+    )
+    with TestClient(api.app) as client:
+        yield client
+
+
+@pytest.fixture
+def core_client() -> DummyCoreClient:
+    return DummyCoreClient()
+
+
+@pytest.fixture
+def transactions_client() -> DummyTransactionsClient:
+    return DummyTransactionsClient()
+
+
+@pytest.fixture
+def item_collection(item: Item) -> ItemCollection:
+    return {
+        "type": "FeatureCollection",
+        "features": [item],
+        "links": [],
+        "context": None,
+    }
+
+
+@pytest.fixture
+def item() -> Item:
+    return {
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "stac_extensions": [],
+        "id": "test_item",
+        "geometry": {"type": "Point", "coordinates": [-105, 40]},
+        "bbox": [-105, 40, -105, 40],
+        "properties": {},
+        "links": [],
+        "assets": {},
+        "collection": "test_collection",
+    }

stac_fastapi/types/stac_fastapi/types/core.py

@@ -28,18 +28,21 @@ class BaseTransactionsClient(abc.ABC):
 
     @abc.abstractmethod
     def create_item(
-        self, collection_id: str, item: stac_types.Item, **kwargs
-    ) -> Optional[Union[stac_types.Item, Response]]:
+        self,
+        collection_id: str,
+        item: Union[stac_types.Item, stac_types.ItemCollection],
+        **kwargs,
+    ) -> Optional[Union[stac_types.Item, Response, None]]:
         """Create a new item.
 
         Called with `POST /collections/{collection_id}/items`.
 
         Args:
-            item: the item
+            item: the item or item collection
             collection_id: the id of the collection from the resource path
 
         Returns:
-            The item that was created.
+            The item that was created or None if item collection.
 
         """
         ...
@@ -138,17 +141,21 @@ class AsyncBaseTransactionsClient(abc.ABC):
 
     @abc.abstractmethod
     async def create_item(
-        self, collection_id: str, item: stac_types.Item, **kwargs
-    ) -> Optional[Union[stac_types.Item, Response]]:
+        self,
+        collection_id: str,
+        item: Union[stac_types.Item, stac_types.ItemCollection],
+        **kwargs,
+    ) -> Optional[Union[stac_types.Item, Response, None]]:
         """Create a new item.
 
         Called with `POST /collections/{collection_id}/items`.
 
         Args:
-            item: the item
+            item: the item or item collection
+            collection_id: the id of the collection from the resource path
 
         Returns:
-            The item that was created.
+            The item that was created or None if item collection.
 
         """
         ...

stac_fastapi/types/stac_fastapi/types/stac.py

@@ -1,6 +1,6 @@
 """STAC types."""
 import sys
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Union
 
 # Avoids a Pydantic error:
 # TypeError: You should use `typing_extensions.TypedDict` instead of `typing.TypedDict` with Python < 3.9.2.
@@ -58,7 +58,7 @@ class Collection(Catalog, total=False):
 class Item(TypedDict, total=False):
     """STAC Item."""
 
-    type: str
+    type: Literal["Feature"]
     stac_version: str
     stac_extensions: Optional[List[str]]
     id: str
@@ -73,7 +73,7 @@ class Item(TypedDict, total=False):
 class ItemCollection(TypedDict, total=False):
     """STAC Item Collection."""
 
-    type: str
+    type: Literal["FeatureCollection"]
     features: List[Item]
     links: List[Dict[str, Any]]
     context: Optional[Dict[str, int]]