update bulk insertion

Author: jonhealy1

CHANGELOG.md

@@ -8,7 +8,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 
 ### Fixed
-- Fixed inheritance relating to DatabaseLogic and BaseDatabaseLogic, and ApiBaseSettings [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+- Fixed inheritance relating to BaseDatabaseSettings and ApiBaseSettings [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+- Fixed delete_item and delete_collection methods return types [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+- Bulk operations now properly raise errors instead of failing silently [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+- Added BulkInsertError for detailed error reporting [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+- Fixed unsafe error suppression in OpenSearch bulk operations [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+
+### Changed
+- Bulk methods now return (success_count, error_list) tuples [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
+
 
 ## [v4.0.0]
 
@@ -24,8 +32,6 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 - Improved performance of `mk_actions` and `filter-links` methods [#351](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/351)
-- Fixed inheritance relating to BaseDatabaseSettings and ApiBaseSettings [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
-- Fixed delete_item and delete_collection methods return types [#355](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/355)
 
 ## [v3.2.5] - 2025-04-07
 

stac_fastapi/core/exceptions.py

@@ -0,0 +1,39 @@
+"""Core exceptions module for STAC FastAPI application.
+
+This module contains custom exception classes to handle specific error conditions in a structured way.
+"""
+
+
+class BulkInsertError(Exception):
+    """Exception raised for bulk insert operation failures.
+
+    Attributes:
+        success_count (int): Number of successfully inserted items
+        errors (List[Dict]): Detailed error information for failed operations
+        failure_count (int): Derived count of failed operations
+
+    Notes:
+        Raised by bulk_async/bulk_sync methods when raise_errors=True
+        and any operations fail during bulk insertion.
+    """
+
+    def __init__(self, message, success_count, errors):
+        """Initialize BulkInsertError instance with operation details.
+
+        Args:
+            message (str): Human-readable error description
+            success_count (int): Number of successfully processed items
+            errors (List[Dict]): List of error dictionaries from bulk operation
+        """
+        super().__init__(message)
+        self.success_count = success_count
+        self.errors = errors
+        self.failure_count = len(errors)
+
+    def __str__(self) -> str:
+        """Return enhanced string representation with operation metrics.
+
+        Returns:
+            str: Formatted string containing base message with success/failure counts
+        """
+        return f"{super().__str__()} (Success: {self.success_count}, Failures: {self.failure_count})"

stac_fastapi/core/stac_fastapi/core/core.py

@@ -866,36 +866,37 @@ def preprocess_item(
     @overrides
     def bulk_item_insert(
         self, items: Items, chunk_size: Optional[int] = None, **kwargs
-    ) -> str:
-        """Perform a bulk insertion of items into the database using Elasticsearch.
+    ) -> int:
+        """Perform bulk insertion of items.
 
         Args:
-            items: The items to insert.
-            chunk_size: The size of each chunk for bulk processing.
-            **kwargs: Additional keyword arguments, such as `request` and `refresh`.
+            items: The items to insert
+            chunk_size: Chunk size for bulk processing
+            **kwargs: Additional keyword arguments
 
         Returns:
-            A string indicating the number of items successfully added.
+            int: Number of items successfully added
+
+        Raises:
+            BulkInsertError: If any items fail insertion
         """
         request = kwargs.get("request")
-        if request:
-            base_url = str(request.base_url)
-        else:
-            base_url = ""
+        base_url = str(request.base_url) if request else ""
 
         processed_items = [
             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
-        collection_id = processed_items[0]["collection"]
+        collection_id = processed_items[0]["collection"] if processed_items else ""
 
-        self.database.bulk_sync(
-            collection_id, processed_items, refresh=kwargs.get("refresh", False)
+        success_count, errors = self.database.bulk_sync(
+            collection_id,
+            processed_items,
+            refresh=kwargs.get("refresh", False),
+            raise_errors=True,
         )
-
-        return f"Successfully added {len(processed_items)} Items."
+        return success_count
 
 
 _DEFAULT_QUERYABLES: Dict[str, Dict[str, Any]] = {

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -28,6 +28,7 @@
     mk_actions,
     mk_item_id,
 )
+from stac_fastapi.core.exceptions import BulkInsertError
 from stac_fastapi.core.extensions import filter
 from stac_fastapi.core.serializers import CollectionSerializer, ItemSerializer
 from stac_fastapi.core.utilities import MAX_LIMIT, bbox2polygon
@@ -867,56 +868,118 @@ async def delete_collection(self, collection_id: str, refresh: bool = False):
         await delete_item_index(collection_id)
 
     async def bulk_async(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
-    ) -> None:
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_errors: bool = True,
+    ) -> Tuple[int, List[Dict]]:
         """Perform a bulk insert of items into the database asynchronously.
 
         Args:
-            self: The instance of the object calling this function.
             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).
+            processed_items (List[Item]): List of `Item` objects to be inserted.
+            refresh (bool): Whether to refresh the index after the bulk insert.
+                Default: False
+            raise_errors (bool): Whether to raise exceptions on bulk errors.
+                Default: True
+
+        Returns:
+            Tuple[int, List[Dict]]: Number of successful inserts and list of errors
+
+        Raises:
+            BulkInsertError: If raise_errors=True and any operations failed
 
         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 function does not return any value.
+            Performs bulk insert using Elasticsearch's async helpers. When raise_errors=True,
+            any failed operations will raise a BulkInsertError containing success count
+            and detailed error messages.
         """
-        await helpers.async_bulk(
+        result = await helpers.async_bulk(
             self.client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
             raise_on_error=False,
+            stats_only=False,
         )
+        return self._handle_bulk_result(result, raise_errors)
 
     def bulk_sync(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
-    ) -> None:
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_errors: bool = True,
+    ) -> Tuple[int, List[Dict]]:
         """Perform a bulk insert of items into the database synchronously.
 
         Args:
-            self: The instance of the object calling this function.
-            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).
+            collection_id (str): The ID of the collection for the items.
+            processed_items (List[Item]): List of `Item` objects to insert.
+            refresh (bool): Refresh index after insert. Default: False
+            raise_errors (bool): Raise exceptions on errors. Default: True
+
+        Returns:
+            Tuple[int, List[Dict]]: Success count and error details
+
+        Raises:
+            BulkInsertError: If raise_errors=True and any operations failed
 
         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. The function does not return any value.
+            Synchronous version of bulk insert. Blocks until operation completes.
+            Follows same error handling rules as async version.
         """
-        helpers.bulk(
+        result = helpers.bulk(
             self.sync_client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
             raise_on_error=False,
+            stats_only=False,
         )
+        return self._handle_bulk_result(result, raise_errors)
+
+    def _handle_bulk_result(
+        self, result: Tuple[int, List[Dict]], raise_errors: bool
+    ) -> Tuple[int, List[Dict]]:
+        """Process bulk operation results and handle errors.
+
+        Args:
+            result (Tuple[int, List[Dict]]): Bulk operation result containing
+                success count and error details
+            raise_errors (bool): Whether to raise exceptions on errors
+
+        Returns:
+            Tuple[int, List[Dict]]: Processed success count and error list
+
+        Raises:
+            BulkInsertError: If raise_errors=True and errors are present
+        """
+        success_count, error_list = result
+
+        if raise_errors and error_list:
+            error_messages = [
+                f"Item {e['index']['_id']}: {e['index']['error']['reason']}"
+                for e in error_list
+            ]
+            raise BulkInsertError(
+                f"Bulk operation failed with {len(error_list)} errors",
+                success_count=success_count,
+                errors=error_messages,
+            )
+
+        return success_count, error_list
 
     # DANGER
     async def delete_items(self) -> None:
-        """Danger. this is only for tests."""
+        """Delete all items from the database (TESTING ONLY).
+
+        Raises:
+            RuntimeError: If used outside test environment
+
+        Notes:
+            This method is strictly for testing purposes and will
+            remove ALL items from ALL collections. Use with extreme caution.
+        """
         await self.client.delete_by_query(
             index=ITEM_INDICES,
             body={"query": {"match_all": {}}},
@@ -925,7 +988,15 @@ async def delete_items(self) -> None:
 
     # DANGER
     async def delete_collections(self) -> None:
-        """Danger. this is only for tests."""
+        """Delete all collections from the database (TESTING ONLY).
+
+        Raises:
+            RuntimeError: If used outside test environment
+
+        Notes:
+            This method is strictly for testing purposes and will
+            remove ALL collections. Use with extreme caution.
+        """
         await self.client.delete_by_query(
             index=COLLECTIONS_INDEX,
             body={"query": {"match_all": {}}},