update os db logic

Author: jonhealy1

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -31,7 +31,7 @@
 )
 from stac_fastapi.core.extensions import filter
 from stac_fastapi.core.serializers import CollectionSerializer, ItemSerializer
-from stac_fastapi.core.utilities import MAX_LIMIT, bbox2polygon
+from stac_fastapi.core.utilities import MAX_LIMIT, bbox2polygon, resolve_refresh
 from stac_fastapi.opensearch.config import (
     AsyncOpensearchSettings as AsyncSearchSettings,
 )
@@ -864,15 +864,17 @@ def bulk_sync_prep_create_item(
     async def create_item(
         self,
         item: Item,
-        refresh: bool = False,
         base_url: str = "",
         exist_ok: bool = False,
+        **kwargs: Any,
     ):
         """Database logic for creating one item.
 
         Args:
             item (Item): The item to be created.
-            refresh (bool, optional): Refresh the index after performing the operation. Defaults to False.
+            base_url (str, optional): The base URL for the item. Defaults to an empty string.
+            exist_ok (bool, optional): Whether to allow the item to exist already. Defaults to False.
+            **kwargs: Additional keyword arguments like refresh.
 
         Raises:
             ConflictError: If the item already exists in the database.
@@ -883,6 +885,19 @@ async def create_item(
         # todo: check if collection exists, but cache
         item_id = item["id"]
         collection_id = item["collection"]
+
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the creation attempt
+        logger.info(
+            f"Creating item {item_id} in collection {collection_id} with refresh={refresh}"
+        )
+
         item = await self.async_prep_create_item(
             item=item, base_url=base_url, exist_ok=exist_ok
         )
@@ -893,19 +908,29 @@ async def create_item(
             refresh=refresh,
         )
 
-    async def delete_item(
-        self, item_id: str, collection_id: str, refresh: bool = False
-    ):
+    async def delete_item(self, item_id: str, collection_id: str, **kwargs: Any):
         """Delete a single item from the database.
 
         Args:
             item_id (str): The id of the Item to be deleted.
             collection_id (str): The id of the Collection that the Item belongs to.
-            refresh (bool, optional): Whether to refresh the index after the deletion. Default is False.
+            **kwargs: Additional keyword arguments like refresh.
 
         Raises:
             NotFoundError: If the Item does not exist in the database.
         """
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the deletion attempt
+        logger.info(
+            f"Deleting item {item_id} from collection {collection_id} with refresh={refresh}"
+        )
+
         try:
             await self.client.delete(
                 index=index_alias_by_collection_id(collection_id),
@@ -935,12 +960,12 @@ async def get_items_mapping(self, collection_id: str) -> Dict[str, Any]:
         except exceptions.NotFoundError:
             raise NotFoundError(f"Mapping for index {index_name} not found")
 
-    async def create_collection(self, collection: Collection, refresh: bool = False):
+    async def create_collection(self, collection: Collection, **kwargs: Any):
         """Create a single collection in the database.
 
         Args:
             collection (Collection): The Collection object to be created.
-            refresh (bool, optional): Whether to refresh the index after the creation. Default is False.
+            **kwargs: Additional keyword arguments like refresh.
 
         Raises:
             ConflictError: If a Collection with the same id already exists in the database.
@@ -950,6 +975,16 @@ async def create_collection(self, collection: Collection, refresh: bool = False)
         """
         collection_id = collection["id"]
 
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the creation attempt
+        logger.info(f"Creating collection {collection_id} with refresh={refresh}")
+
         if await self.client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise ConflictError(f"Collection {collection_id} already exists")
 
@@ -989,14 +1024,14 @@ async def find_collection(self, collection_id: str) -> Collection:
         return collection["_source"]
 
     async def update_collection(
-        self, collection_id: str, collection: Collection, refresh: bool = False
+        self, collection_id: str, collection: Collection, **kwargs: Any
     ):
         """Update a collection from the database.
 
         Args:
-            self: The instance of the object calling this function.
             collection_id (str): The ID of the collection to be updated.
             collection (Collection): The Collection object to be used for the update.
+            **kwargs: Additional keyword arguments like refresh.
 
         Raises:
             NotFoundError: If the collection with the given `collection_id` is not
@@ -1007,9 +1042,23 @@ async def update_collection(
             `collection_id` and with the collection specified in the `Collection` object.
             If the collection is not found, a `NotFoundError` is raised.
         """
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the update attempt
+        logger.info(f"Updating collection {collection_id} with refresh={refresh}")
+
         await self.find_collection(collection_id=collection_id)
 
         if collection_id != collection["id"]:
+            logger.info(
+                f"Collection ID change detected: {collection_id} -> {collection['id']}"
+            )
+
             await self.create_collection(collection, refresh=refresh)
 
             await self.client.reindex(
@@ -1025,7 +1074,7 @@ async def update_collection(
                 refresh=refresh,
             )
 
-            await self.delete_collection(collection_id)
+            await self.delete_collection(collection_id=collection_id, **kwargs)
 
         else:
             await self.client.index(
@@ -1035,23 +1084,34 @@ async def update_collection(
                 refresh=refresh,
             )
 
-    async def delete_collection(self, collection_id: str, refresh: bool = False):
+    async def delete_collection(self, collection_id: str, **kwargs: Any):
         """Delete a collection from the database.
 
         Parameters:
             self: The instance of the object calling this function.
             collection_id (str): The ID of the collection to be deleted.
-            refresh (bool): Whether to refresh the index after the deletion (default: False).
+            **kwargs: Additional keyword arguments like refresh.
 
         Raises:
             NotFoundError: If the collection with the given `collection_id` is not found in the database.
 
         Notes:
             This function first verifies that the collection with the specified `collection_id` exists in the database, and then
-            deletes the collection. If `refresh` is set to True, the index is refreshed after the deletion. Additionally, this
-            function also calls `delete_item_index` to delete the index for the items in the collection.
+            deletes the collection. If `refresh` is set to "true", "false", or "wait_for", the index is refreshed accordingly after
+            the deletion. Additionally, this function also calls `delete_item_index` to delete the index for the items in the collection.
         """
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
         await self.find_collection(collection_id=collection_id)
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the deletion attempt
+        logger.info(f"Deleting collection {collection_id} with refresh={refresh}")
+
         await self.client.delete(
             index=COLLECTIONS_INDEX, id=collection_id, refresh=refresh
         )
@@ -1061,15 +1121,17 @@ async def bulk_async(
         self,
         collection_id: str,
         processed_items: List[Item],
-        refresh: bool = False,
+        **kwargs: Any,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database asynchronously.
 
         Args:
             collection_id (str): The ID of the collection to which the items belong.
             processed_items (List[Item]): A list of `Item` objects to be inserted into the database.
-            refresh (bool): Whether to refresh the index after the bulk insert (default: False).
+            **kwargs (Any): Additional keyword arguments, including:
+                - refresh (str, optional): Whether to refresh the index after the bulk insert.
+                Can be "true", "false", or "wait_for". Defaults to the value of `self.sync_settings.database_refresh`.
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1078,32 +1140,58 @@ async def bulk_async(
 
         Notes:
             This function performs a bulk insert of `processed_items` into the database using the specified `collection_id`.
-            The insert is performed asynchronously, and the event loop is used to run the operation in a separate executor.
-            The `mk_actions` function is called to generate a list of actions for the bulk insert. If `refresh` is set to True,
-            the index is refreshed after the bulk insert.
+            The insert is performed synchronously and blocking, meaning that the function does not return until the insert has
+            completed. The `mk_actions` function is called to generate a list of actions for the bulk insert. The `refresh`
+            parameter determines whether the index is refreshed after the bulk insert:
+                - "true": Forces an immediate refresh of the index.
+                - "false": Does not refresh the index immediately (default behavior).
+                - "wait_for": Waits for the next refresh cycle to make the changes visible.
         """
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.async_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the bulk insert attempt
+        logger.info(
+            f"Performing bulk insert for collection {collection_id} with refresh={refresh}"
+        )
+
+        # Handle empty processed_items
+        if not processed_items:
+            logger.warning(f"No items to insert for collection {collection_id}")
+            return 0, []
+
         raise_on_error = self.async_settings.raise_on_bulk_error
         success, errors = await helpers.async_bulk(
             self.client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
             raise_on_error=raise_on_error,
         )
+        # Log the result
+        logger.info(
+            f"Bulk insert completed for collection {collection_id}: {success} successes, {len(errors)} errors"
+        )
         return success, errors
 
     def bulk_sync(
         self,
         collection_id: str,
         processed_items: List[Item],
-        refresh: bool = False,
+        **kwargs: Any,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database synchronously.
 
         Args:
             collection_id (str): The ID of the collection to which the items belong.
             processed_items (List[Item]): A list of `Item` objects to be inserted into the database.
-            refresh (bool): Whether to refresh the index after the bulk insert (default: False).
+            **kwargs (Any): Additional keyword arguments, including:
+                - refresh (str, optional): Whether to refresh the index after the bulk insert.
+                Can be "true", "false", or "wait_for". Defaults to the value of `self.sync_settings.database_refresh`.
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1113,9 +1201,29 @@ def bulk_sync(
         Notes:
             This function performs a bulk insert of `processed_items` into the database using the specified `collection_id`.
             The insert is performed synchronously and blocking, meaning that the function does not return until the insert has
-            completed. The `mk_actions` function is called to generate a list of actions for the bulk insert. If `refresh` is set to
-            True, the index is refreshed after the bulk insert.
+            completed. The `mk_actions` function is called to generate a list of actions for the bulk insert. The `refresh`
+            parameter determines whether the index is refreshed after the bulk insert:
+                - "true": Forces an immediate refresh of the index.
+                - "false": Does not refresh the index immediately (default behavior).
+                - "wait_for": Waits for the next refresh cycle to make the changes visible.
         """
+        # Ensure kwargs is a dictionary
+        kwargs = kwargs or {}
+
+        # Resolve the `refresh` parameter
+        refresh = kwargs.get("refresh", self.sync_settings.database_refresh)
+        refresh = resolve_refresh(str(refresh).lower())
+
+        # Log the bulk insert attempt
+        logger.info(
+            f"Performing bulk insert for collection {collection_id} with refresh={refresh}"
+        )
+
+        # Handle empty processed_items
+        if not processed_items:
+            logger.warning(f"No items to insert for collection {collection_id}")
+            return 0, []
+
         raise_on_error = self.sync_settings.raise_on_bulk_error
         success, errors = helpers.bulk(
             self.sync_client,