upgrade stac-fastapi to v2.4.9, implement support for BulkTransaction method parameter

Author: StijnCaerts

stac_fastapi/elasticsearch/setup.py

@@ -10,9 +10,9 @@
     "attrs",
     "pydantic[dotenv]<2",
     "stac_pydantic==2.0.*",
-    "stac-fastapi.types==2.4.8",
-    "stac-fastapi.api==2.4.8",
-    "stac-fastapi.extensions==2.4.8",
+    "stac-fastapi.types==2.4.9",
+    "stac-fastapi.api==2.4.9",
+    "stac-fastapi.extensions==2.4.9",
     "elasticsearch[async]==7.17.9",
     "elasticsearch-dsl==7.4.1",
     "pystac[validation]",

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -26,7 +26,7 @@
 from stac_fastapi.elasticsearch.session import Session
 from stac_fastapi.extensions.third_party.bulk_transactions import (
     BaseBulkTransactionsClient,
-    Items,
+    Items, BulkTransactionMethod,
 )
 from stac_fastapi.types import stac as stac_types
 from stac_fastapi.types.config import Settings
@@ -568,7 +568,7 @@ async def create_item(
         if item["type"] == "FeatureCollection":
             bulk_client = BulkTransactionsClient()
             processed_items = [
-                bulk_client.preprocess_item(item, base_url) for item in item["features"]  # type: ignore
+                bulk_client.preprocess_item(item, base_url, BulkTransactionMethod.INSERT) for item in item["features"]  # type: ignore
             ]
 
             await self.database.bulk_async(
@@ -718,17 +718,19 @@ def __attrs_post_init__(self):
         settings = ElasticsearchSettings()
         self.client = settings.create_client
 
-    def preprocess_item(self, item: stac_types.Item, base_url) -> stac_types.Item:
+    def preprocess_item(self, item: stac_types.Item, base_url, method: BulkTransactionMethod) -> stac_types.Item:
         """Preprocess an item to match the data model.
 
         Args:
             item: The item to preprocess.
             base_url: The base URL of the request.
+            method: The bulk transaction method.
 
         Returns:
             The preprocessed item.
         """
-        return self.database.sync_prep_create_item(item=item, base_url=base_url)
+        exist_ok = (method == BulkTransactionMethod.UPSERT)
+        return self.database.sync_prep_create_item(item=item, base_url=base_url, exist_ok=exist_ok)
 
     @overrides
     def bulk_item_insert(
@@ -751,7 +753,7 @@ def bulk_item_insert(
             base_url = ""
 
         processed_items = [
-            self.preprocess_item(item, base_url) for item in items.items.values()
+            self.preprocess_item(item, base_url, items.method) for item in items.items.values()
         ]
 
         # not a great way to get the collection_id-- should be part of the method signature

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -591,13 +591,14 @@ async def check_collection_exists(self, collection_id: str):
         if not await self.client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise NotFoundError(f"Collection {collection_id} does not exist")
 
-    async def prep_create_item(self, item: Item, base_url: str) -> Item:
+    async def prep_create_item(self, item: Item, base_url: str, exist_ok: bool = False) -> Item:
         """
         Preps an item for insertion into the database.
 
         Args:
             item (Item): The item to be prepped for insertion.
             base_url (str): The base URL used to create the item's self URL.
+            exist_ok (bool): Indicates whether the item can exist already.
 
         Returns:
             Item: The prepped item.
@@ -608,7 +609,7 @@ async def prep_create_item(self, item: Item, base_url: str) -> Item:
         """
         await self.check_collection_exists(collection_id=item["collection"])
 
-        if await self.client.exists(
+        if not exist_ok and await self.client.exists(
             index=index_by_collection_id(item["collection"]),
             id=mk_item_id(item["id"], item["collection"]),
         ):
@@ -618,7 +619,7 @@ async def prep_create_item(self, item: Item, base_url: str) -> Item:
 
         return self.item_serializer.stac_to_db(item, base_url)
 
-    def sync_prep_create_item(self, item: Item, base_url: str) -> Item:
+    def sync_prep_create_item(self, item: Item, base_url: str, exist_ok: bool = False) -> Item:
         """
         Prepare an item for insertion into the database.
 
@@ -629,6 +630,7 @@ def sync_prep_create_item(self, item: Item, base_url: str) -> Item:
         Args:
             item (Item): The item to be inserted into the database.
             base_url (str): The base URL used for constructing URLs for the item.
+            exist_ok (bool): Indicates whether the item can exist already.
 
         Returns:
             Item: The item after preparation is done.
@@ -642,7 +644,7 @@ def sync_prep_create_item(self, item: Item, base_url: str) -> Item:
         if not self.sync_client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise NotFoundError(f"Collection {collection_id} does not exist")
 
-        if self.sync_client.exists(
+        if not exist_ok and self.sync_client.exists(
             index=index_by_collection_id(collection_id),
             id=mk_item_id(item_id, collection_id),
         ):