Merge branch 'main' into stac_to_db

Author: jonhealy1

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/transactions.py

@@ -53,7 +53,7 @@ def _create_item_index(self):
 
     def create_item(self, model: stac_types.Item, **kwargs):
         """Create item."""
-        base_url = str(kwargs["request"].base_url)
+        base_url = str(kwargs["request"].base_url
 
         self._create_item_index()
 
@@ -87,6 +87,7 @@ def create_collection(self, model: stac_types.Collection, **kwargs):
         self.client.index(
             index="stac_collections", doc_type="_doc", id=model["id"], document=model
         )
+        return CollectionSerializer.db_to_stac(model, base_url)
 
     def update_item(self, model: stac_types.Item, **kwargs):
         """Update item."""
@@ -159,6 +160,13 @@ def _preprocess_item(self, model: stac_types.Item, base_url) -> stac_types.Item:
         item = ItemSerializer.stac_to_db(model, base_url)
         return item
 
+    def bulk_sync(self, processed_items):
+        """Elasticsearch bulk insertion."""
+        actions = [
+            {"_index": "stac_items", "_source": item} for item in processed_items
+        ]
+        helpers.bulk(self.client, actions)
+
     def bulk_item_insert(self, items: Items, **kwargs) -> str:
         """Bulk item insertion using es."""
         transactions_client = TransactionsClient()
@@ -172,13 +180,6 @@ def bulk_item_insert(self, items: Items, **kwargs) -> str:
         ]
         return_msg = f"Successfully added {len(processed_items)} items."
 
-        def bulk_sync(processed_items):
-            actions = [
-                {"_index": "stac_items", "_source": item} for item in processed_items
-            ]
-
-            helpers.bulk(self.client, actions)
-
-        bulk_sync(processed_items)
+        self.bulk_sync(processed_items)
 
         return return_msg

stac_fastapi/elasticsearch/tests/clients/test_elasticsearch.py

@@ -290,6 +290,31 @@ def test_bulk_item_insert(
     #     )
 
 
+def test_feature_collection_insert(
+    es_core: CoreCrudClient,
+    es_transactions: TransactionsClient,
+    es_bulk_transactions: BulkTransactionsClient,
+    load_test_data: Callable,
+):
+    coll = load_test_data("test_collection.json")
+    es_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}
+
+    es_transactions.create_item(feature_collection, request=MockStarletteRequest)
+    time.sleep(3)
+    fc = es_core.item_collection(coll["id"], request=MockStarletteRequest)
+    assert len(fc["features"]) >= 10
+
+
 @pytest.mark.skip(reason="Not working")
 def test_landing_page_no_collection_title(
     es_core: CoreCrudClient,