added fixes in insertion and selection strategies

Grzegorz Pustulka · Grzegorz Pustulka · commit 0c2d8e6ac5ae · 2025-07-02T00:52:33.000+02:00
diff --git a/Makefile b/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 --cov=stac_fastapi --cov-report=term-missing'
+	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest -s --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 --cov=stac_fastapi --cov-report=term-missing'
+	-$(run_os) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest -s --cov=stac_fastapi --cov-report=term-missing'
 	docker compose down
 
 .PHONY: run-database-es
diff --git a/compose.yml b/compose.yml
@@ -21,6 +21,9 @@ services:
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
       - BACKEND=elasticsearch
+      - ENABLE_DATETIME_INDEX_FILTERING=true
+      - DATETIME_INDEX_MAX_SIZE_GB=0
+      - DATETIME_INDEX_MAX_SIZE_GB=0.00002
     ports:
       - "8080:8080"
     volumes:
@@ -55,6 +58,8 @@ services:
       - ES_VERIFY_CERTS=false
       - BACKEND=opensearch
       - STAC_FASTAPI_RATE_LIMIT=200/minute
+      - ENABLE_DATETIME_INDEX_FILTERING=true
+      - DATETIME_INDEX_MAX_SIZE_GB=0.00002
     ports:
       - "8082:8082"
     volumes:
diff --git a/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/__init__.py b/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/__init__.py
@@ -42,8 +42,16 @@
 )
 from .index_insertion_strategies import (
     AsyncIndexInserter,
+    AsyncSimpleIndexInsertion,
+    BaseIndexInserter,
+    ElasticsearchAdapter,
     IndexInsertionFactory,
+    OpenSearchAdapter,
+    SearchEngineAdapter,
+    SearchEngineAdapterFactory,
+    SearchEngineType,
     SyncIndexInserter,
+    SyncSimpleIndexInsertion,
 )
 from .index_selection_strategies import (
     AsyncDatetimeBasedIndexSelector,
@@ -76,8 +84,16 @@
     "AsyncDatetimeBasedIndexSelector",
     # Index insertion strategies
     "AsyncIndexInserter",
-    "SyncIndexInserter",
+    "AsyncSimpleIndexInsertion",
+    "BaseIndexInserter",
+    "ElasticsearchAdapter",
     "IndexInsertionFactory",
+    "OpenSearchAdapter",
+    "SearchEngineAdapter",
+    "SearchEngineAdapterFactory",
+    "SearchEngineType",
+    "SyncIndexInserter",
+    "SyncSimpleIndexInsertion",
     # Query operations
     "apply_free_text_filter_shared",
     "apply_intersects_filter_shared",
@@ -93,4 +109,4 @@
     "return_date",
     "extract_date",
     "extract_date_from_index",
-]
+]
diff --git a/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/datetime.py b/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/datetime.py
@@ -72,6 +72,7 @@ def extract_date(date_str: str) -> date:
     Returns:
         A date object extracted from the input string.
     """
+    date_str = date_str.replace('Z', '+00:00')
     return datetime_type.fromisoformat(date_str).date()
 
 
diff --git a/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index_insertion_strategies.py b/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index_insertion_strategies.py
@@ -54,13 +54,13 @@ def create_datetime_index_sync(
     ) -> str:
         pass
 
-    async def create_index_alias(self, client: Any, collection_id: str, end_date: str):
+    async def update_index_alias(self, client: Any, collection_id: str, end_date: str):
         index = index_alias_by_collection_id(collection_id)
         await client.indices.put_alias(
             index=index, name=self.alias_by_index_and_end_date(index, end_date)
         )
 
-    def create_index_alias_sync(
+    def update_index_alias_sync(
         self, sync_client: Any, collection_id: str, end_date: str
     ):
         index = index_alias_by_collection_id(collection_id)
@@ -91,11 +91,12 @@ async def create_datetime_index(
         self, client: Any, collection_id: str, start_date: str
     ) -> str:
         index_name = self.index_by_collection_id_and_date(collection_id, start_date)
+        alias_name = index_name.removeprefix(ITEMS_INDEX_PREFIX)
         await client.options(ignore_status=400).indices.create(
             index=index_name,
-            body={"aliases": {index_alias_by_collection_id(collection_id): {}}},
+            body={"aliases": {index_alias_by_collection_id(collection_id): {}, alias_name: {}}},
         )
-        return index_name
+        return alias_name
 
     def create_simple_index_sync(self, sync_client: Any, collection_id: str) -> str:
         index_name = f"{index_by_collection_id(collection_id)}-000001"
@@ -109,11 +110,12 @@ def create_datetime_index_sync(
         self, sync_client: Any, collection_id: str, start_date: str
     ) -> str:
         index_name = self.index_by_collection_id_and_date(collection_id, start_date)
+        alias_name = index_name.removeprefix(ITEMS_INDEX_PREFIX)
         sync_client.options(ignore_status=400).indices.create(
             index=index_name,
-            body={"aliases": {index_alias_by_collection_id(collection_id): {}}},
+            body={"aliases": {index_alias_by_collection_id(collection_id): {}, alias_name: {}}},
         )
-        return index_name
+        return alias_name
 
 
 class OpenSearchAdapter(SearchEngineAdapter):
@@ -135,15 +137,16 @@ async def create_datetime_index(
         self, client: Any, collection_id: str, start_date: str
     ) -> str:
         index_name = self.index_by_collection_id_and_date(collection_id, start_date)
+        alias_name = index_name.removeprefix(ITEMS_INDEX_PREFIX)
         await client.indices.create(
             index=index_name,
             body={
-                "aliases": {index_alias_by_collection_id(collection_id): {}},
+                "aliases": {index_alias_by_collection_id(collection_id): {}, alias_name: {}},
                 "mappings": ES_ITEMS_MAPPINGS,
                 "settings": ES_ITEMS_SETTINGS,
             },
         )
-        return index_name
+        return alias_name
 
     def create_simple_index_sync(self, sync_client: Any, collection_id: str) -> str:
         index_name = f"{index_by_collection_id(collection_id)}-000001"
@@ -163,15 +166,16 @@ def create_datetime_index_sync(
         self, sync_client: Any, collection_id: str, start_date: str
     ) -> str:
         index_name = self.index_by_collection_id_and_date(collection_id, start_date)
+        alias_name = index_name.removeprefix(ITEMS_INDEX_PREFIX)
         sync_client.indices.create(
             index=index_name,
             body={
-                "aliases": {index_alias_by_collection_id(collection_id): {}},
+                "aliases": {index_alias_by_collection_id(collection_id): {}, alias_name: {}},
                 "mappings": ES_ITEMS_MAPPINGS,
                 "settings": ES_ITEMS_SETTINGS,
             },
         )
-        return index_name
+        return alias_name
 
 
 class SearchEngineAdapterFactory:
@@ -217,6 +221,10 @@ def should_create_collection_index(self) -> bool:
     async def create_simple_index(self, client: Any, collection_id: str):
         return await self.search_adapter.create_simple_index(client, collection_id)
 
+    async def get_index_size_in_gb(self, index_name: str) -> float:
+        data = await self.client.indices.stats(index=index_name)
+        return data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
+
     async def _get_target_index_base(
         self,
         index_selector,
@@ -243,16 +251,14 @@ async def _get_target_index_base(
             return target_index
 
         if check_size:
-            data = await self.client.indices.stats(index=target_index)
-            current_size_in_gb = (
-                data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
-            )
+            breakpoint()
+            index_size_gb = await self.get_index_size_in_gb(target_index)
             max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", 20))
 
-            if current_size_in_gb > max_size_gb:
+            if index_size_gb > max_size_gb:
                 end_date = extract_date(product_datetime)
                 if end_date != extract_date_from_index(all_indexes[-1]):
-                    await self.search_adapter.create_index_alias(
+                    await self.search_adapter.update_index_alias(
                         self.client, collection_id, str(end_date)
                     )
                     target_index = await self.search_adapter.create_datetime_index(
@@ -298,18 +304,15 @@ async def prepare_bulk_actions(
         new_index = None
 
         if first_item_index == latest_index:
-            data = await self.client.indices.stats(index=first_item_index)
-            current_size_in_gb = (
-                data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
-            )
+            index_size_gb = await self.get_index_size_in_gb(first_item_index)
             max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", 20))
 
-            if current_size_in_gb > max_size_gb:
+            if index_size_gb > max_size_gb:
                 current_index_end_date = extract_date_from_index(first_item_index)
                 first_item_date = extract_date(first_item["properties"]["datetime"])
 
                 if first_item_date != current_index_end_date:
-                    await self.search_adapter.create_index_alias(
+                    await self.search_adapter.update_index_alias(
                         self.client, collection_id, str(current_index_end_date)
                     )
                     next_day_start = current_index_end_date + timedelta(days=1)
@@ -348,6 +351,10 @@ def __init__(self, sync_client, search_adapter: SearchEngineAdapter):
     def should_create_collection_index(self) -> bool:
         return False
 
+    def get_index_size_in_gb(self, index_name: str) -> float:
+        data = self.sync_client.indices.stats(index=index_name)
+        return data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
+
     def _get_target_index_base(
         self,
         index_selector,
@@ -372,16 +379,13 @@ def _get_target_index_base(
             return target_index
 
         if check_size:
-            data = self.sync_client.indices.stats(index=target_index)
-            current_size_in_gb = (
-                data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
-            )
+            index_size_gb = self.get_index_size_in_gb(target_index)
             max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", 20))
 
-            if current_size_in_gb > max_size_gb:
+            if index_size_gb > max_size_gb:
                 end_date = extract_date(product_datetime)
                 if end_date != extract_date_from_index(all_indexes[-1]):
-                    self.search_adapter.create_index_alias_sync(
+                    self.search_adapter.update_index_alias_sync(
                         self.sync_client, collection_id, str(end_date)
                     )
                     target_index = self.search_adapter.create_datetime_index_sync(
@@ -425,18 +429,14 @@ def prepare_bulk_actions(
         new_index = None
 
         if first_item_index == latest_index:
-            data = self.sync_client.indices.stats(index=first_item_index)
-            current_size_in_gb = (
-                data["_all"]["primaries"]["store"]["size_in_bytes"] / 1e9
-            )
+            index_size_gb = self.get_index_size_in_gb(first_item_index)
             max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", 20))
-
-            if current_size_in_gb > max_size_gb:
+            if index_size_gb > max_size_gb:
                 current_index_end_date = extract_date_from_index(first_item_index)
                 first_item_date = extract_date(first_item["properties"]["datetime"])
 
                 if first_item_date != current_index_end_date:
-                    self.search_adapter.create_index_alias_sync(
+                    self.search_adapter.update_index_alias_sync(
                         self.sync_client, collection_id, str(current_index_end_date)
                     )
                     next_day_start = current_index_end_date + timedelta(days=1)
diff --git a/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index_selection_strategies.py b/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index_selection_strategies.py
@@ -69,16 +69,25 @@ async def select_indexes(
 class AsyncDatetimeBasedIndexSelector(IndexSelectionStrategy):
     """Asynchronous index selector that filters indices based on datetime criteria with caching."""
 
+    _instance = None
+
+    def __new__(cls, client):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
     def __init__(self, client):
         """Initialize the datetime-based index selector.
 
         Args:
             client: Elasticsearch/OpenSearch client instance used for querying
                 index aliases and metadata.
         """
-        self.client = client
-        self._aliases_cache: Optional[Dict[str, List[str]]] = None
-        self._cache_timestamp: float = 0
+        if not hasattr(self, '_initialized'):
+            self.client = client
+            self._aliases_cache: Optional[Dict[str, List[str]]] = None
+            self._cache_timestamp: float = 0
+            self._initialized = True
 
     @property
     def _cache_expired(self) -> bool:
@@ -101,14 +110,13 @@ async def _load_aliases_cache(self) -> Dict[str, List[str]]:
         """
         response = await self.client.indices.get_alias(index=f"{ITEMS_INDEX_PREFIX}*")
         result = {}
-
         for index_info in response.values():
             aliases = index_info.get("aliases", {})
             base_alias = None
             items_aliases = []
 
             for alias_name in aliases.keys():
-                if alias_name.startswith(ITEMS_INDEX_PREFIX):
+                if not alias_name.startswith(ITEMS_INDEX_PREFIX):
                     items_aliases.append(alias_name)
                 else:
                     base_alias = alias_name
@@ -151,7 +159,7 @@ async def get_collection_indexes(self, collection_id: str) -> List[str]:
                 Returns empty list if collection is not found in cache.
         """
         cache = await self.get_aliases_cache()
-        return cache.get(collection_id, [])
+        return cache.get(index_alias_by_collection_id(collection_id), [])
 
     async def select_indexes(
         self,
@@ -194,16 +202,25 @@ async def select_indexes(
 class SyncDatetimeBasedIndexSelector(IndexSelectionStrategy):
     """Synchronous index selector that filters indices based on datetime criteria with caching."""
 
+    _instance = None
+
+    def __new__(cls, client):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
     def __init__(self, sync_client):
         """Initialize the datetime-based index selector.
 
         Args:
             sync_client: Synchronous Elasticsearch/OpenSearch client instance used for querying
                 index aliases and metadata.
         """
-        self.sync_client = sync_client
-        self._aliases_cache: Optional[Dict[str, List[str]]] = None
-        self._cache_timestamp: float = 0
+        if not hasattr(self, '_initialized'):
+            self.sync_client = sync_client
+            self._aliases_cache: Optional[Dict[str, List[str]]] = None
+            self._cache_timestamp: float = 0
+            self._initialized = True
 
     @property
     def _cache_expired(self) -> bool:
@@ -233,7 +250,7 @@ def _load_aliases_cache(self) -> Dict[str, List[str]]:
             items_aliases = []
 
             for alias_name in aliases.keys():
-                if alias_name.startswith(ITEMS_INDEX_PREFIX):
+                if not alias_name.startswith(ITEMS_INDEX_PREFIX):
                     items_aliases.append(alias_name)
                 else:
                     base_alias = alias_name
@@ -276,7 +293,7 @@ def get_collection_indexes(self, collection_id: str) -> List[str]:
                 Returns empty list if collection is not found in cache.
         """
         cache = self.get_aliases_cache()
-        return cache.get(collection_id, [])
+        return cache.get(index_alias_by_collection_id(collection_id), [])
 
     def select_indexes(
         self,
diff --git a/stac_fastapi/tests/api/test_api.py b/stac_fastapi/tests/api/test_api.py
diff --git a/stac_fastapi/tests/index/test_index_insertion.py b/stac_fastapi/tests/index/test_index_insertion.py
