Merge branch 'main' of github.com:stac-utils/stac-fastapi-elasticsearch into landing_page_id

Author: rhysrevans3

CHANGELOG.md

@@ -11,12 +11,32 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - Added configurable landing page ID `STAC_FASTAPI_LANDING_PAGE_ID` [#352](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/352)
 - Updated dynamic mapping for items to map long values to double versus float [#326](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/326)
+
+### Changed
+
+### Fixed
+
+## [v4.1.0] - 2025-05-09
+
+### Added
+
+- Added logging to bulk insertion methods to provide detailed feedback on errors encountered during operations. [#364](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/364)
+- Introduced the `RAISE_ON_BULK_ERROR` environment variable to control whether bulk insertion methods raise exceptions on errors (`true`) or log warnings and continue processing (`false`). [#364](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/364)
+- Added code coverage reporting to the test suite using pytest-cov. [#87](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/issues/87)
+
+### Changed
+
+- Updated dynamic mapping for items to map long values to double versus float. [#326](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/326)
 - Extended Datetime Search to search on start_datetime and end_datetime as well as datetime fields. [#182](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/182)
+- Changed item update operation to use Elasticsearch index API instead of delete and create for better efficiency and atomicity. [#75](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/issues/75)
+- Bulk insertion via `BulkTransactionsClient` now strictly validates all STAC Items using the Pydantic model before insertion. Any invalid item will immediately raise a `ValidationError`, ensuring consistent validation with single-item inserts and preventing invalid STAC Items from being stored. This validation is enforced regardless of the `RAISE_ON_BULK_ERROR` setting. [#368](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/368)
 
 ### Changed
 
 ### Fixed
 
+- Refactored `create_item` and `update_item` methods to share unified logic, ensuring consistent conflict detection, validation, and database operations. [#368](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/368)
+
 ## [v4.0.0] - 2025-04-23
 
 ### Added
@@ -352,7 +372,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Use genexp in execute_search and get_all_collections to return results.
 - Added db_to_stac serializer to item_collection method in core.py.
 
-[Unreleased]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v4.0.0...main
+[Unreleased]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v4.1.0...main
+[v4.1.0]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v4.0.0...v4.1.0
 [v4.0.0]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v3.2.5...v4.0.0
 [v3.2.5]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v3.2.4...v3.2.5
 [v3.2.4]: https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/compare/v3.2.3...v3.2.4

Makefile

@@ -75,10 +75,10 @@ test-opensearch:
 
 .PHONY: test
 test:
-	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest'
+	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest --cov=stac_fastapi --cov-report=term-missing'
 	docker compose down
 
-	-$(run_os) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest'
+	-$(run_os) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest --cov=stac_fastapi --cov-report=term-missing'
 	docker compose down
 
 .PHONY: run-database-es

README.md

@@ -114,7 +114,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. **Note:** STAC Item and ItemCollection validation errors will always raise, regardless of this flag. | `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.

compose.yml

@@ -9,7 +9,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-elasticsearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Elasticsearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - STAC_FASTAPI_LANDING_PAGE_ID=stac-fastapi-elasticsearch
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -42,7 +42,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-opensearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Opensearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - APP_HOST=0.0.0.0
       - APP_PORT=8082
       - RELOAD=true

examples/auth/compose.basic_auth.yml

@@ -9,7 +9,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-elasticsearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Elasticsearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - STAC_FASTAPI_LANDING_PAGE_ID=stac-fastapi-elasticsearch
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -43,7 +43,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-opensearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Opensearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - APP_HOST=0.0.0.0
       - APP_PORT=8082
       - RELOAD=true

examples/auth/compose.oauth2.yml

@@ -9,7 +9,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-elasticsearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Elasticsearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - STAC_FASTAPI_LANDING_PAGE_ID=stac-fastapi-elasticsearch
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -44,7 +44,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-opensearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Opensearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - APP_HOST=0.0.0.0
       - APP_PORT=8082
       - RELOAD=true

examples/auth/compose.route_dependencies.yml

@@ -9,7 +9,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-elasticsearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Elasticsearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - STAC_FASTAPI_LANDING_PAGE_ID=stac-fastapi-elasticsearch
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -43,7 +43,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-opensearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Opensearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - APP_HOST=0.0.0.0
       - APP_PORT=8082
       - RELOAD=true

examples/rate_limit/compose.rate_limit.yml

@@ -9,7 +9,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-elasticsearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Elasticsearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - STAC_FASTAPI_LANDING_PAGE_ID=stac-fastapi-elasticsearch
       - APP_HOST=0.0.0.0
       - APP_PORT=8080
@@ -43,7 +43,7 @@ services:
     environment:
       - STAC_FASTAPI_TITLE=stac-fastapi-opensearch
       - STAC_FASTAPI_DESCRIPTION=A STAC FastAPI with an Opensearch backend
-      - STAC_FASTAPI_VERSION=4.0.0
+      - STAC_FASTAPI_VERSION=4.1.0
       - APP_HOST=0.0.0.0
       - APP_PORT=8082
       - RELOAD=true

stac_fastapi/core/stac_fastapi/core/core.py

@@ -676,46 +676,65 @@ class TransactionsClient(AsyncBaseTransactionsClient):
     @overrides
     async def create_item(
         self, collection_id: str, item: Union[Item, ItemCollection], **kwargs
-    ) -> Optional[stac_types.Item]:
-        """Create an item in the collection.
+    ) -> Union[stac_types.Item, str]:
+        """
+        Create an item or a feature collection of items in the specified collection.
 
         Args:
-            collection_id (str): The id of the collection to add the item to.
-            item (stac_types.Item): The item to be added to the collection.
-            kwargs: Additional keyword arguments.
+            collection_id (str): The ID of the collection to add the item(s) to.
+            item (Union[Item, ItemCollection]): A single item or a collection of items to be added.
+            **kwargs: Additional keyword arguments, such as `request` and `refresh`.
 
         Returns:
-            stac_types.Item: The created item.
+            Union[stac_types.Item, str]: The created item if a single item is added, or a summary string
+            indicating the number of items successfully added and errors if a collection of items is added.
 
         Raises:
-            NotFound: If the specified collection is not found in the database.
-            ConflictError: If the item in the specified collection already exists.
-
+            NotFoundError: If the specified collection is not found in the database.
+            ConflictError: If an item with the same ID already exists in the collection.
         """
-        item = item.model_dump(mode="json")
-        base_url = str(kwargs["request"].base_url)
+        request = kwargs.get("request")
+        base_url = str(request.base_url)
 
-        # If a feature collection is posted
-        if item["type"] == "FeatureCollection":
+        # Convert Pydantic model to dict for uniform processing
+        item_dict = item.model_dump(mode="json")
+
+        # Handle FeatureCollection (bulk insert)
+        if item_dict["type"] == "FeatureCollection":
             bulk_client = BulkTransactionsClient(
                 database=self.database, settings=self.settings
             )
+            features = item_dict["features"]
             processed_items = [
                 bulk_client.preprocess_item(
-                    item, base_url, BulkTransactionMethod.INSERT
+                    feature, base_url, BulkTransactionMethod.INSERT
                 )
-                for item in item["features"]
+                for feature in features
             ]
-
-            await self.database.bulk_async(
-                collection_id, processed_items, refresh=kwargs.get("refresh", False)
+            attempted = len(processed_items)
+            success, errors = await self.database.bulk_async(
+                collection_id,
+                processed_items,
+                refresh=kwargs.get("refresh", False),
             )
+            if errors:
+                logger.error(
+                    f"Bulk async operation encountered errors for collection {collection_id}: {errors} (attempted {attempted})"
+                )
+            else:
+                logger.info(
+                    f"Bulk async operation succeeded with {success} actions for collection {collection_id}."
+                )
+            return f"Successfully added {success} Items. {attempted - success} errors occurred."
 
-            return None
-        else:
-            item = await self.database.prep_create_item(item=item, base_url=base_url)
-            await self.database.create_item(item, refresh=kwargs.get("refresh", False))
-            return ItemSerializer.db_to_stac(item, base_url)
+        # Handle single item
+        await self.database.create_item(
+            item_dict,
+            refresh=kwargs.get("refresh", False),
+            base_url=base_url,
+            exist_ok=False,
+        )
+        return ItemSerializer.db_to_stac(item_dict, base_url)
 
     @overrides
     async def update_item(
@@ -741,9 +760,9 @@ async def update_item(
         now = datetime_type.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
         item["properties"]["updated"] = now
 
-        await self.database.check_collection_exists(collection_id)
-        await self.delete_item(item_id=item_id, collection_id=collection_id)
-        await self.create_item(collection_id=collection_id, item=Item(**item), **kwargs)
+        await self.database.create_item(
+            item, refresh=kwargs.get("refresh", False), base_url=base_url, exist_ok=True
+        )
 
         return ItemSerializer.db_to_stac(item, base_url)
 
@@ -876,7 +895,7 @@ def preprocess_item(
             The preprocessed item.
         """
         exist_ok = method == BulkTransactionMethod.UPSERT
-        return self.database.sync_prep_create_item(
+        return self.database.bulk_sync_prep_create_item(
             item=item, base_url=base_url, exist_ok=exist_ok
         )
 
@@ -900,19 +919,32 @@ def bulk_item_insert(
         else:
             base_url = ""
 
-        processed_items = [
-            self.preprocess_item(item, base_url, items.method)
-            for item in items.items.values()
-        ]
+        processed_items = []
+        for item in items.items.values():
+            try:
+                validated = Item(**item) if not isinstance(item, Item) else item
+                processed_items.append(
+                    self.preprocess_item(
+                        validated.model_dump(mode="json"), base_url, items.method
+                    )
+                )
+            except ValidationError:
+                # Immediately raise on the first invalid item (strict mode)
+                raise
 
-        # not a great way to get the collection_id-- should be part of the method signature
         collection_id = processed_items[0]["collection"]
-
-        self.database.bulk_sync(
-            collection_id, processed_items, refresh=kwargs.get("refresh", False)
+        attempted = len(processed_items)
+        success, errors = self.database.bulk_sync(
+            collection_id,
+            processed_items,
+            refresh=kwargs.get("refresh", False),
         )
+        if errors:
+            logger.error(f"Bulk sync operation encountered errors: {errors}")
+        else:
+            logger.info(f"Bulk sync operation succeeded with {success} actions.")
 
-        return f"Successfully added {len(processed_items)} Items."
+        return f"Successfully added/updated {success} Items. {attempted - success} errors occurred."
 
 
 _DEFAULT_QUERYABLES: Dict[str, Dict[str, Any]] = {

stac_fastapi/core/stac_fastapi/core/version.py

@@ -1,2 +1,2 @@
 """library version."""
-__version__ = "4.0.0"
+__version__ = "4.1.0"