part_1 - sync versions removed

Author: Grzegorz Pustulka

compose.yml

@@ -3,6 +3,7 @@ services:
     container_name: stac-fastapi-es
     image: stac-utils/stac-fastapi-es
     restart: always
+    platform: linux/amd64
     build:
       context: .
       dockerfile: dockerfiles/Dockerfile.dev.es
@@ -21,6 +22,7 @@ services:
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
       - BACKEND=elasticsearch
+      - ENABLE_DATETIME_INDEX_FILTERING=true
     ports:
       - "8080:8080"
     volumes:
@@ -36,6 +38,7 @@ services:
     container_name: stac-fastapi-os
     image: stac-utils/stac-fastapi-os
     restart: always
+    platform: linux/amd64
     build:
       context: .
       dockerfile: dockerfiles/Dockerfile.dev.os
@@ -70,6 +73,7 @@ services:
     container_name: es-container
     image: docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION:-8.11.0}
     hostname: elasticsearch
+    platform: linux/amd64
     environment:
       ES_JAVA_OPTS: -Xms512m -Xmx1g
       action.destructive_requires_name: false
@@ -83,6 +87,7 @@ services:
     container_name: os-container
     image: opensearchproject/opensearch:${OPENSEARCH_VERSION:-2.11.1}
     hostname: opensearch
+    platform: linux/amd64
     environment:
       - discovery.type=single-node
       - plugins.security.disabled=true

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -27,7 +27,6 @@
     PartialItem,
     PatchOperation,
 )
-from stac_fastapi.sfeos_helpers import filter
 from stac_fastapi.sfeos_helpers.database import (
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
@@ -36,6 +35,7 @@
     get_queryables_mapping_shared,
     index_alias_by_collection_id,
     index_by_collection_id,
+    mk_actions,
     mk_item_id,
     populate_sort_shared,
     validate_refresh,
@@ -44,6 +44,7 @@
     merge_to_operations,
     operations_to_script,
 )
+from stac_fastapi.sfeos_helpers.filter import cql2 as filter
 from stac_fastapi.sfeos_helpers.mappings import (
     AGGREGATION_MAPPING,
     COLLECTIONS_INDEX,
@@ -54,8 +55,8 @@
 )
 from stac_fastapi.sfeos_helpers.search_engine import (
     BaseIndexInserter,
+    BaseIndexSelector,
     IndexInsertionFactory,
-    IndexSelectionStrategy,
     IndexSelectorFactory,
 )
 from stac_fastapi.types.errors import ConflictError, NotFoundError
@@ -137,10 +138,8 @@ class DatabaseLogic(BaseDatabaseLogic):
     sync_settings: SyncElasticsearchSettings = attr.ib(
         factory=SyncElasticsearchSettings
     )
-    async_index_selector: IndexSelectionStrategy = attr.ib(init=False)
-    sync_index_selector: IndexSelectionStrategy = attr.ib(init=False)
+    async_index_selector: BaseIndexSelector = attr.ib(init=False)
     async_index_inserter: BaseIndexInserter = attr.ib(init=False)
-    sync_index_inserter: BaseIndexInserter = attr.ib(init=False)
 
     client = attr.ib(init=False)
     sync_client = attr.ib(init=False)
@@ -149,18 +148,10 @@ def __attrs_post_init__(self):
         """Initialize clients after the class is instantiated."""
         self.client = self.async_settings.create_client
         self.sync_client = self.sync_settings.create_client
-        self.async_index_inserter = (
-            IndexInsertionFactory.create_async_insertion_strategy(self.client)
-        )
-        self.sync_index_inserter = IndexInsertionFactory.create_sync_insertion_strategy(
-            self.sync_client
-        )
-        self.async_index_selector = IndexSelectorFactory.create_async_selector(
+        self.async_index_inserter = IndexInsertionFactory.create_insertion_strategy(
             self.client
         )
-        self.sync_index_selector = IndexSelectorFactory.create_sync_selector(
-            self.sync_client
-        )
+        self.async_index_selector = IndexSelectorFactory.create_selector(self.client)
 
     item_serializer: Type[ItemSerializer] = attr.ib(default=ItemSerializer)
     collection_serializer: Type[CollectionSerializer] = attr.ib(
@@ -1448,12 +1439,9 @@ def bulk_sync(
 
         # Perform the bulk insert
         raise_on_error = self.sync_settings.raise_on_bulk_error
-        actions = self.sync_index_inserter.prepare_bulk_actions(
-            collection_id, processed_items
-        )
         success, errors = helpers.bulk(
             self.sync_client,
-            actions,
+            mk_actions(collection_id, processed_items),
             refresh=refresh,
             raise_on_error=raise_on_error,
         )

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -26,14 +26,14 @@
     AsyncOpensearchSettings as AsyncSearchSettings,
 )
 from stac_fastapi.opensearch.config import OpensearchSettings as SyncSearchSettings
-from stac_fastapi.sfeos_helpers import filter
 from stac_fastapi.sfeos_helpers.database import (
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     create_index_templates_shared,
     delete_item_index_shared,
     get_queryables_mapping_shared,
     index_alias_by_collection_id,
+    mk_actions,
     mk_item_id,
     populate_sort_shared,
     validate_refresh,
@@ -42,6 +42,7 @@
     merge_to_operations,
     operations_to_script,
 )
+from stac_fastapi.sfeos_helpers.filter import cql2 as filter
 from stac_fastapi.sfeos_helpers.mappings import (
     AGGREGATION_MAPPING,
     COLLECTIONS_INDEX,
@@ -53,8 +54,8 @@
 )
 from stac_fastapi.sfeos_helpers.search_engine import (
     BaseIndexInserter,
+    BaseIndexSelector,
     IndexInsertionFactory,
-    IndexSelectionStrategy,
     IndexSelectorFactory,
 )
 from stac_fastapi.types.errors import ConflictError, NotFoundError
@@ -120,10 +121,8 @@ class DatabaseLogic(BaseDatabaseLogic):
     async_settings: AsyncSearchSettings = attr.ib(factory=AsyncSearchSettings)
     sync_settings: SyncSearchSettings = attr.ib(factory=SyncSearchSettings)
 
-    async_index_selector: IndexSelectionStrategy = attr.ib(init=False)
-    sync_index_selector: IndexSelectionStrategy = attr.ib(init=False)
+    async_index_selector: BaseIndexSelector = attr.ib(init=False)
     async_index_inserter: BaseIndexInserter = attr.ib(init=False)
-    sync_index_inserter: BaseIndexInserter = attr.ib(init=False)
 
     client = attr.ib(init=False)
     sync_client = attr.ib(init=False)
@@ -132,18 +131,10 @@ def __attrs_post_init__(self):
         """Initialize clients after the class is instantiated."""
         self.client = self.async_settings.create_client
         self.sync_client = self.sync_settings.create_client
-        self.async_index_inserter = (
-            IndexInsertionFactory.create_async_insertion_strategy(self.client)
-        )
-        self.sync_index_inserter = IndexInsertionFactory.create_sync_insertion_strategy(
-            self.sync_client
-        )
-        self.async_index_selector = IndexSelectorFactory.create_async_selector(
+        self.async_index_inserter = IndexInsertionFactory.create_insertion_strategy(
             self.client
         )
-        self.sync_index_selector = IndexSelectorFactory.create_sync_selector(
-            self.sync_client
-        )
+        self.async_index_selector = IndexSelectorFactory.create_selector(self.client)
 
     item_serializer: Type[ItemSerializer] = attr.ib(default=ItemSerializer)
     collection_serializer: Type[CollectionSerializer] = attr.ib(
@@ -1323,6 +1314,7 @@ async def bulk_async(
                 - "false": Does not refresh the index immediately (default behavior).
                 - "wait_for": Waits for the next refresh cycle to make the changes visible.
         """
+        breakpoint()
         # Ensure kwargs is a dictionary
         kwargs = kwargs or {}
 
@@ -1407,14 +1399,15 @@ def bulk_sync(
             logger.warning(f"No items to insert for collection {collection_id}")
             return 0, []
 
-        raise_on_error = self.sync_settings.raise_on_bulk_error
-        actions = self.sync_index_inserter.prepare_bulk_actions(
-            collection_id, processed_items
-        )
+        # 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,
-            actions,
+            mk_actions(collection_id, processed_items),
             refresh=refresh,
             raise_on_error=raise_on_error,
         )

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/__init__.py

@@ -31,7 +31,7 @@
 
 # Re-export all functions for backward compatibility
 from .datetime import extract_date, extract_first_date_from_index, return_date
-from .document import mk_item_id
+from .document import mk_actions, mk_item_id
 from .index import (
     create_index_templates_shared,
     delete_item_index_shared,
@@ -64,6 +64,7 @@
     "get_queryables_mapping_shared",
     # Document operations
     "mk_item_id",
+    "mk_actions",
     # Utility functions
     "validate_refresh",
     "get_bool_env",

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/document.py

@@ -4,6 +4,11 @@
 including document ID generation and bulk action creation.
 """
 
+from typing import Any, Dict, List
+
+from stac_fastapi.sfeos_helpers.database.index import index_alias_by_collection_id
+from stac_fastapi.types.stac import Item
+
 
 def mk_item_id(item_id: str, collection_id: str) -> str:
     """Create the document id for an Item in Elasticsearch.
@@ -16,3 +21,28 @@ def mk_item_id(item_id: str, collection_id: str) -> str:
         str: The document id for the Item, combining the Item id and the Collection id, separated by a `|` character.
     """
     return f"{item_id}|{collection_id}"
+
+
+def mk_actions(collection_id: str, processed_items: List[Item]) -> List[Dict[str, Any]]:
+    """Create Elasticsearch bulk actions for a list of processed items.
+
+    Args:
+        collection_id (str): The identifier for the collection the items belong to.
+        processed_items (List[Item]): The list of processed items to be bulk indexed.
+
+    Returns:
+        List[Dict[str, Union[str, Dict]]]: The list of bulk actions to be executed,
+        each action being a dictionary with the following keys:
+        - `_index`: the index to store the document in.
+        - `_id`: the document's identifier.
+        - `_source`: the source of the document.
+    """
+    index_alias = index_alias_by_collection_id(collection_id)
+    return [
+        {
+            "_index": index_alias,
+            "_id": mk_item_id(item["id"], item["collection"]),
+            "_source": item,
+        }
+        for item in processed_items
+    ]

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/__init__.py

@@ -1,33 +1,27 @@
 """Search engine index management package."""
 
-from .async_inserters import AsyncDatetimeIndexInserter, AsyncSimpleIndexInserter
-from .base import BaseAsyncIndexInserter, BaseSyncIndexInserter
+from .base import BaseIndexInserter
 from .factory import IndexInsertionFactory
 from .index_operations import IndexOperations
+from .inserters import DatetimeIndexInserter, SimpleIndexInserter
 from .managers import DatetimeIndexManager, IndexSizeManager
 from .selection import (
-    AsyncDatetimeBasedIndexSelector,
-    IndexSelectionStrategy,
+    BaseIndexSelector,
+    DatetimeBasedIndexSelector,
     IndexSelectorFactory,
-    SyncDatetimeBasedIndexSelector,
     UnfilteredIndexSelector,
 )
-from .sync_inserters import SyncDatetimeIndexInserter, SyncSimpleIndexInserter
 
 __all__ = [
-    "BaseAsyncIndexInserter",
-    "BaseSyncIndexInserter",
+    "BaseIndexInserter",
+    "BaseIndexSelector",
+    "IndexOperations",
     "IndexSizeManager",
     "DatetimeIndexManager",
-    "AsyncDatetimeIndexInserter",
-    "AsyncSimpleIndexInserter",
-    "SyncDatetimeIndexInserter",
-    "SyncSimpleIndexInserter",
-    "IndexOperations",
+    "DatetimeIndexInserter",
+    "SimpleIndexInserter",
     "IndexInsertionFactory",
-    "IndexSelectionStrategy",
-    "AsyncDatetimeBasedIndexSelector",
-    "SyncDatetimeBasedIndexSelector",
+    "DatetimeBasedIndexSelector",
     "UnfilteredIndexSelector",
     "IndexSelectorFactory",
 ]

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/base.py

@@ -4,7 +4,7 @@
 from typing import Any, Dict, List
 
 
-class BaseAsyncIndexInserter(ABC):
+class BaseIndexInserter(ABC):
     """Base async index inserter with common async methods."""
 
     @abstractmethod
@@ -49,48 +49,3 @@ async def create_simple_index(self, client: Any, collection_id: str) -> str:
             str: Created index name.
         """
         pass
-
-
-class BaseSyncIndexInserter(ABC):
-    """Base sync index inserter with common sync methods."""
-
-    @abstractmethod
-    def get_target_index(self, collection_id: str, product: Dict[str, Any]) -> str:
-        """Get target index for a product synchronously.
-
-        Args:
-            collection_id (str): Collection identifier.
-            product (Dict[str, Any]): Product data.
-
-        Returns:
-            str: Target index name.
-        """
-        pass
-
-    @abstractmethod
-    def prepare_bulk_actions(
-        self, collection_id: str, items: List[Dict[str, Any]]
-    ) -> List[Dict[str, Any]]:
-        """Prepare bulk actions for multiple items synchronously.
-
-        Args:
-            collection_id (str): Collection identifier.
-            items (List[Dict[str, Any]]): List of items to process.
-
-        Returns:
-            List[Dict[str, Any]]: List of bulk actions.
-        """
-        pass
-
-    @abstractmethod
-    def create_simple_index(self, client: Any, collection_id: str) -> str:
-        """Create a simple index synchronously.
-
-        Args:
-            client: Search engine client instance.
-            collection_id (str): Collection identifier.
-
-        Returns:
-            str: Created index name.
-        """
-        pass