bulk raise on error env var

Author: jonhealy1

README.md

@@ -113,7 +113,8 @@ You can customize additional settings in your `.env` file:
 | `BACKEND`                    | Tests-related variable                                                               | `elasticsearch` or `opensearch` based on the backend | Optional                                                                                    |
 | `ELASTICSEARCH_VERSION`          | Version of Elasticsearch to use.                                                         | `8.11.0`                      | Optional                                                                                    |
 | `ENABLE_DIRECT_RESPONSE`         | Enable direct response for maximum performance (disables all FastAPI dependencies, including authentication, custom status codes, and validation) | `false`                  | Optional                                                                                    |
-| `OPENSEARCH_VERSION`         | OpenSearch version                                                                   | `2.11.1`                 | Optional                                                                                    |
+| `OPENSEARCH_VERSION`         | OpenSearch version                                                                   | `2.11.1`                 | Optional       
+| `RAISE_ON_BULK_ERROR`         | Controls whether bulk insert operations raise exceptions on errors. If set to `true`, the operation will stop and raise an exception when an error occurs. If set to `false`, errors will be logged, and the operation will continue. | `false`                  | Optional                                                                                |
 
 > [!NOTE]
 > The variables `ES_HOST`, `ES_PORT`, `ES_USE_SSL`, and `ES_VERIFY_CERTS` apply to both Elasticsearch and OpenSearch backends, so there is no need to rename the key names to `OS_` even if you're using OpenSearch.

stac_fastapi/core/stac_fastapi/core/core.py

@@ -1,6 +1,7 @@
 """Core client."""
 
 import logging
+import os
 from collections import deque
 from datetime import datetime as datetime_type
 from datetime import timezone
@@ -709,7 +710,10 @@ async def create_item(
             ]
             attempted = len(processed_items)
             success, errors = await self.database.bulk_async(
-                collection_id, processed_items, refresh=kwargs.get("refresh", False)
+                collection_id,
+                processed_items,
+                refresh=kwargs.get("refresh", False),
+                raise_on_error=os.getenv("RAISE_ON_BULK_ERROR", False),
             )
             if errors:
                 logger.error(f"Bulk async operation encountered errors: {errors}")
@@ -914,7 +918,10 @@ def bulk_item_insert(
         collection_id = processed_items[0]["collection"]
         attempted = len(processed_items)
         success, errors = self.database.bulk_sync(
-            collection_id, processed_items, refresh=kwargs.get("refresh", False)
+            collection_id,
+            processed_items,
+            refresh=kwargs.get("refresh", False),
+            raise_on_error=os.getenv("RAISE_ON_BULK_ERROR", False),
         )
         if errors:
             logger.error(f"Bulk sync operation encountered errors: {errors}")

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -985,7 +985,11 @@ 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
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_on_error: bool = False,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database asynchronously.
@@ -994,6 +998,7 @@ async def bulk_async(
             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).
+            raise_on_error (bool): Whether to raise an error if the bulk operation fails (default: False).
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1010,12 +1015,16 @@ async def bulk_async(
             self.client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
-            raise_on_error=False,  # Do not raise errors
+            raise_on_error=raise_on_error,
         )
         return success, errors
 
     def bulk_sync(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_on_error: bool = False,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database synchronously.
@@ -1024,6 +1033,7 @@ def bulk_sync(
             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).
+            raise_on_error (bool): Whether to raise an error if the bulk operation fails (default: False).
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1040,7 +1050,7 @@ def bulk_sync(
             self.sync_client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
-            raise_on_error=False,  # Do not raise errors
+            raise_on_error=raise_on_error,
         )
         return success, errors
 

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -983,7 +983,11 @@ 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
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_on_error: bool = False,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database asynchronously.
@@ -992,6 +996,7 @@ async def bulk_async(
             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).
+            raise_on_error (bool): Whether to raise an error if the bulk operation fails (default: False).
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1008,12 +1013,16 @@ async def bulk_async(
             self.client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
-            raise_on_error=False,  # Do not raise errors
+            raise_on_error=raise_on_error,
         )
         return success, errors
 
     def bulk_sync(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
+        self,
+        collection_id: str,
+        processed_items: List[Item],
+        refresh: bool = False,
+        raise_on_error: bool = False,
     ) -> Tuple[int, List[Dict[str, Any]]]:
         """
         Perform a bulk insert of items into the database synchronously.
@@ -1022,6 +1031,7 @@ def bulk_sync(
             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).
+            raise_on_error (bool): Whether to raise an error if the bulk operation fails (default: False).
 
         Returns:
             Tuple[int, List[Dict[str, Any]]]: A tuple containing:
@@ -1038,7 +1048,7 @@ def bulk_sync(
             self.sync_client,
             mk_actions(collection_id, processed_items),
             refresh=refresh,
-            raise_on_error=False,  # Do not raise errors
+            raise_on_error=raise_on_error,
         )
         return success, errors