POST a feature collection to the items route (#367)

* process feature collection

* add insert feature colllection test

* tests pass

* run pre-commit

* clean up test

* run pre-commit again

* Update changelog

* return none

* pre-commit

* Update return type of create_item function

Co-authored-by: Nathan Zimmerman <[email protected]>

Author: jonhealy1

CHANGES.md

@@ -5,6 +5,7 @@
 ### Added
 
 * Add hook to allow adding dependencies to routes. ([#295](https://github.com/stac-utils/stac-fastapi/pull/295))
+* Ability to POST an ItemCollection to the collections/{collectionId}/items route. ([#367](https://github.com/stac-utils/stac-fastapi/pull/367))
 * Add STAC API - Collections conformance class. ([383](https://github.com/stac-utils/stac-fastapi/pull/383))
 
 ### Changed

stac_fastapi/sqlalchemy/stac_fastapi/sqlalchemy/transactions.py

@@ -35,11 +35,19 @@ class TransactionsClient(BaseTransactionsClient):
     )
 
     def create_item(
-        self, item: stac_types.Item, **kwargs
-    ) -> Optional[Union[stac_types.Item, Response]]:
+        self, model: Union[stac_types.Item, stac_types.ItemCollection], **kwargs
+    ) -> Optional[stac_types.Item]:
         """Create item."""
         base_url = str(kwargs["request"].base_url)
-        data = self.item_serializer.stac_to_db(item)
+
+        # If a feature collection is posted
+        if model["type"] == "FeatureCollection":
+            bulk_client = BulkTransactionsClient(session=self.session)
+            bulk_client.bulk_item_insert(items=model["features"])
+            return None
+
+        # Otherwise a single item has been posted
+        data = self.item_serializer.stac_to_db(model)
         with self.session.writer.context_session() as session:
             session.add(data)
             return self.item_serializer.db_to_stac(data, base_url)

stac_fastapi/sqlalchemy/tests/clients/test_postgres.py

@@ -312,6 +312,35 @@ def test_bulk_item_insert_chunked(
         )
 
 
+def test_feature_collection_insert(
+    postgres_core: CoreCrudClient,
+    postgres_transactions: TransactionsClient,
+    load_test_data: Callable,
+):
+    coll = load_test_data("test_collection.json")
+    postgres_transactions.create_collection(coll, request=MockStarletteRequest)
+
+    item = load_test_data("test_item.json")
+
+    features = []
+    for _ in range(10):
+        _item = deepcopy(item)
+        _item["id"] = str(uuid.uuid4())
+        features.append(_item)
+
+    feature_collection = {"type": "FeatureCollection", "features": features}
+
+    postgres_transactions.create_item(feature_collection, request=MockStarletteRequest)
+
+    fc = postgres_core.item_collection(coll["id"], request=MockStarletteRequest)
+    assert len(fc["features"]) >= 10
+
+    for item in features:
+        postgres_transactions.delete_item(
+            item["id"], item["collection"], request=MockStarletteRequest
+        )
+
+
 def test_landing_page_no_collection_title(
     postgres_core: CoreCrudClient,
     postgres_transactions: TransactionsClient,