docs

Author: Grzegorz Pustulka

README.md

@@ -225,7 +225,7 @@ You can customize additional settings in your `.env` file:
 | `DATABASE_REFRESH`              | Controls whether database operations refresh the index immediately after changes. If set to `true`, changes will be immediately searchable. If set to `false`, changes may not be immediately visible but can improve performance for bulk operations. If set to `wait_for`, changes will wait for the next refresh cycle to become visible. | `false`                                              | Optional |
 | `ENABLE_TRANSACTIONS_EXTENSIONS` | Enables or disables the Transactions and Bulk Transactions API extensions. If set to `false`, the POST `/collections` route and related transaction endpoints (including bulk transaction operations) will be unavailable in the API. This is useful for deployments where mutating the catalog via the API should be prevented.             | `true`                                               | Optional |
 | `ENABLE_DATETIME_INDEX_FILTERING` | Enable datetime-based index selection using collection IDs. Requires indexes in format: STAC_ITEMS_INDEX_PREFIX_collection-id_start_year-start_month-start_day-end_year-end_month-end_day, e.g. items_sentinel-2-l2a_2025-06-06-2025-09-22.                                                                                                  | `false`                                              | Optional |
-| `DATETIME_INDEX_MAX_SIZE_GB` | Maximum size limit in GB for datetime-based indexes. When an index exceeds this size, a new time-partitioned index will be created. Only applies when `ENABLE_DATETIME_INDEX_FILTERING` is enabled.                                                                                                                                          | `25`                                                 | Optional |
+| `DATETIME_INDEX_MAX_SIZE_GB` | Maximum size limit in GB for datetime-based indexes. When an index exceeds this size, a new time-partitioned index will be created. Note: This value should account for ~25% overhead due to OS/ES caching of data structures and metadata. Only applies when`ENABLE_DATETIME_INDEX_FILTERING` is enabled.                                                                               | `25`                                                 | Optional |
 
 
 > [!NOTE]

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

@@ -4,8 +4,6 @@
 including document ID generation and bulk action creation.
 """
 
-from stac_fastapi.sfeos_helpers.database.index import index_alias_by_collection_id
-
 
 def mk_item_id(item_id: str, collection_id: str) -> str:
     """Create the document id for an Item in Elasticsearch.

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index.py

@@ -7,7 +7,7 @@
 from functools import lru_cache
 from typing import Any, List, Optional
 
-from dateutil.parser import parse
+from dateutil.parser import parse  # type: ignore[import]
 
 from stac_fastapi.sfeos_helpers.mappings import (
     _ES_INDEX_NAME_UNSUPPORTED_CHARS_TABLE,

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/adapters.py

@@ -78,9 +78,8 @@ def create_datetime_index_sync(
         """
         pass
 
-    async def update_index_alias(
-        self, client: Any, end_date: str, old_alias: str
-    ) -> str:
+    @staticmethod
+    async def update_index_alias(client: Any, end_date: str, old_alias: str) -> str:
         """Update index alias with new end date.
 
         Args:
@@ -104,9 +103,8 @@ async def update_index_alias(
         )
         return new_alias
 
-    def update_index_alias_sync(
-        self, client: Any, end_date: str, old_alias: str
-    ) -> str:
+    @staticmethod
+    def update_index_alias_sync(client: Any, end_date: str, old_alias: str) -> str:
         """Update index alias synchronously.
 
         Args:
@@ -236,7 +234,8 @@ def create_datetime_index_sync(
 class OpenSearchAdapter(SearchEngineAdapter):
     """OpenSearch-specific adapter implementation."""
 
-    def _create_index_body(self, aliases: Dict[str, Dict]) -> Dict[str, Any]:
+    @staticmethod
+    def _create_index_body(aliases: Dict[str, Dict]) -> Dict[str, Any]:
         """Create index body with common settings.
 
         Args:

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/async_inserters.py

@@ -96,7 +96,7 @@ async def _get_target_index_internal(
         product: Dict[str, Any],
         check_size: bool = True,
     ) -> str:
-        """Internal method to get target index with size checking.
+        """Get target index with size checking internally.
 
         Args:
             index_selector: Index selector instance.

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

@@ -1,37 +1,39 @@
 """Base classes for index inserters."""
 
 from abc import ABC, abstractmethod
-from typing import Any, Dict, List
+from typing import Any, Awaitable, Dict, List, Union
 
 
 class BaseIndexInserter(ABC):
     """Base class for index insertion strategies."""
 
     @abstractmethod
-    def get_target_index(self, collection_id: str, product: Dict[str, Any]) -> str:
+    def get_target_index(
+        self, collection_id: str, product: Dict[str, Any]
+    ) -> Union[str, Awaitable[str]]:
         """Get target index for a product.
 
         Args:
             collection_id (str): Collection identifier.
             product (Dict[str, Any]): Product data.
 
         Returns:
-            str: Target index name.
+            Union[str, Awaitable[str]]: Target index name or awaitable.
         """
         pass
 
     @abstractmethod
     def prepare_bulk_actions(
         self, collection_id: str, items: List[Dict[str, Any]]
-    ) -> List[Dict[str, Any]]:
+    ) -> Union[List[Dict[str, Any]], Awaitable[List[Dict[str, Any]]]]:
         """Prepare bulk actions for multiple items.
 
         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.
+            Union[List[Dict[str, Any]], Awaitable[List[Dict[str, Any]]]]: List of bulk actions or awaitable.
         """
         pass
 
@@ -78,6 +80,7 @@ async def prepare_bulk_actions(
         """
         pass
 
+    @abstractmethod
     async def create_simple_index(self, client: Any, collection_id: str) -> str:
         """Create a simple index asynchronously.
 
@@ -94,6 +97,35 @@ async def create_simple_index(self, client: Any, collection_id: str) -> str:
 class BaseSyncIndexInserter(BaseIndexInserter):
     """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.
 

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/managers.py

@@ -24,7 +24,7 @@ def __init__(self, client: Any):
             client: Search engine client instance.
         """
         self.client = client
-        self.max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", "20"))
+        self.max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", "25"))
 
     async def get_index_size_in_gb(self, index_name: str) -> float:
         """Get index size in gigabytes asynchronously.
@@ -89,7 +89,8 @@ def __init__(self, client: Any, search_adapter: SearchEngineAdapter):
         self.search_adapter = search_adapter
         self.size_manager = IndexSizeManager(client)
 
-    def _validate_product_datetime(self, product: Dict[str, Any]) -> str:
+    @staticmethod
+    def _validate_product_datetime(product: Dict[str, Any]) -> str:
         """Validate and extract datetime from product.
 
         Args:

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/selection/cache_manager.py

@@ -74,7 +74,7 @@ async def load_aliases(self) -> Dict[str, List[str]]:
             Dict[str, List[str]]: Mapping of base aliases to item aliases.
         """
         response = await self.client.indices.get_alias(index=f"{ITEMS_INDEX_PREFIX}*")
-        result = {}
+        result: Dict[str, List[str]] = {}
 
         for index_info in response.values():
             aliases = index_info.get("aliases", {})
@@ -145,7 +145,7 @@ def load_aliases(self) -> Dict[str, List[str]]:
             Dict[str, List[str]]: Mapping of base aliases to item aliases.
         """
         response = self.client.indices.get_alias(index=f"{ITEMS_INDEX_PREFIX}*")
-        result = {}
+        result: Dict[str, List[str]] = {}
 
         for index_info in response.values():
             aliases = index_info.get("aliases", {})

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/selection/unfiltered_selector.py

@@ -10,7 +10,7 @@
 class UnfilteredIndexSelector(IndexSelectionStrategy):
     """Index selector that returns all available indices without filtering."""
 
-    def select_indexes(
+    async def select_indexes(
         self,
         collection_ids: Optional[List[str]],
         datetime_search: Dict[str, Optional[str]],

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/search_engine/sync_inserters.py

@@ -92,7 +92,7 @@ def _get_target_index_internal(
         product: Dict[str, Any],
         check_size: bool = True,
     ) -> str:
-        """Internal method to get target index with size checking.
+        """Get target index with size checking internally.
 
         Args:
             index_selector: Index selector instance.