bulk sync

Author: jonhealy1

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -5,6 +5,7 @@
 import attr
 import elasticsearch
 from elasticsearch_dsl import Q, Search
+from elasticsearch import helpers
 
 from stac_fastapi.elasticsearch import serializers
 from stac_fastapi.elasticsearch.config import ElasticsearchSettings
@@ -289,3 +290,22 @@ def prep_update_collection(self, collection_id: str):
         except elasticsearch.exceptions.NotFoundError:
             raise NotFoundError(f"Collection {collection_id} not found")
 
+    def delete_collection(self, collection_id: str):
+        try:
+            _ = self.client.get(index=COLLECTIONS_INDEX, id=collection_id)
+        except elasticsearch.exceptions.NotFoundError:
+            raise NotFoundError(f"Collection {collection_id} not found")
+        self.client.delete(index=COLLECTIONS_INDEX, id=collection_id)
+
+     def bulk_sync(self, processed_items):
+        """Database logic for bulk insertion."""
+        actions = [
+            {
+                "_index": ITEMS_INDEX,
+                "_id": mk_item_id(item["id"], item["collection"]),
+                "_source": item,
+            }
+            for item in processed_items
+        ]
+        helpers.bulk(self.client, actions)
+

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/transactions.py

@@ -5,8 +5,6 @@
 from typing import Optional
 
 import attr
-import elasticsearch
-from elasticsearch import helpers
 from overrides import overrides
 
 from stac_fastapi.elasticsearch.config import ElasticsearchSettings
@@ -111,11 +109,7 @@ def update_collection(
     @overrides
     def delete_collection(self, collection_id: str, **kwargs) -> stac_types.Collection:
         """Delete collection."""
-        try:
-            _ = self.client.get(index=COLLECTIONS_INDEX, id=collection_id)
-        except elasticsearch.exceptions.NotFoundError:
-            raise NotFoundError(f"Collection {collection_id} not found")
-        self.client.delete(index=COLLECTIONS_INDEX, id=collection_id)
+        self.database.delete_collection(collection_id=collection_id)
         return None
 
 
@@ -136,18 +130,6 @@ def preprocess_item(self, item: stac_types.Item, base_url) -> stac_types.Item:
         item = self.database.prep_create_item(item=item, base_url=base_url)
         return item
 
-    def bulk_sync(self, processed_items):
-        """Elasticsearch bulk insertion."""
-        actions = [
-            {
-                "_index": ITEMS_INDEX,
-                "_id": mk_item_id(item["id"], item["collection"]),
-                "_source": item,
-            }
-            for item in processed_items
-        ]
-        helpers.bulk(self.client, actions)
-
     @overrides
     def bulk_item_insert(
         self, items: Items, chunk_size: Optional[int] = None, **kwargs
@@ -163,6 +145,6 @@ def bulk_item_insert(
             self.preprocess_item(item, base_url) for item in items.items.values()
         ]
 
-        self.bulk_sync(processed_items)
+        self.database.bulk_sync(processed_items)
 
         return f"Successfully added {len(processed_items)} Items."