added tests for inserts, fixed product deletion, improved time functions, fixed bugs in index_insertion_strategies

Author: Grzegorz Pustulka

compose.yml

@@ -22,8 +22,6 @@ services:
       - 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:
@@ -59,7 +57,6 @@ services:
       - BACKEND=opensearch
       - STAC_FASTAPI_RATE_LIMIT=200/minute
       - ENABLE_DATETIME_INDEX_FILTERING=true
-      - DATETIME_INDEX_MAX_SIZE_GB=0.00002
     ports:
       - "8082:8082"
     volumes:

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -888,10 +888,16 @@ async def delete_item(self, item_id: str, collection_id: str, **kwargs: Any):
 
         try:
             # Perform the delete operation
-            await self.client.delete(
+            await self.client.delete_by_query(
                 index=index_alias_by_collection_id(collection_id),
-                id=mk_item_id(item_id, collection_id),
-                refresh=refresh,
+                body={
+                    "query": {
+                        "term": {
+                            "_id": mk_item_id(item_id, collection_id)
+                        }
+                    }
+                },
+                refresh=refresh
             )
         except ESNotFoundError:
             # Raise a custom NotFoundError if the item does not exist

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -873,10 +873,16 @@ async def delete_item(self, item_id: str, collection_id: str, **kwargs: Any):
         )
 
         try:
-            await self.client.delete(
+            await self.client.delete_by_query(
                 index=index_alias_by_collection_id(collection_id),
-                id=mk_item_id(item_id, collection_id),
-                refresh=refresh,
+                body={
+                    "query": {
+                        "term": {
+                            "_id": mk_item_id(item_id, collection_id)
+                        }
+                    }
+                },
+                refresh=refresh
             )
         except exceptions.NotFoundError:
             raise NotFoundError(

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

@@ -30,7 +30,7 @@
 """
 
 # Re-export all functions for backward compatibility
-from .datetime import extract_date, extract_date_from_index, return_date
+from .datetime import extract_date, return_date, extract_first_date_from_index
 from .document import mk_item_id
 from .index import (
     create_index_templates_shared,
@@ -108,5 +108,5 @@
     # Datetime utilities
     "return_date",
     "extract_date",
-    "extract_date_from_index",
+    "extract_first_date_from_index",
 ]

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/datetime.py

@@ -76,17 +76,19 @@ def extract_date(date_str: str) -> date:
     return datetime_type.fromisoformat(date_str).date()
 
 
-def extract_date_from_index(index: str) -> date:
-    """Extract date from index string containing date pattern.
+def extract_first_date_from_index(index_name: str) -> date:
+    """Extract the first date from an index name containing date patterns.
 
-    Searches for a date pattern (YYYY-MM-DD) within the index string
-    and returns it as a date object.
+    Searches for date patterns (YYYY-MM-DD) within the index name string
+    and returns the first found date as a date object.
 
     Args:
-        index: Index string containing a date pattern.
+        index_name: Index name containing date patterns.
 
     Returns:
-        A date object extracted from the index string.
+        A date object extracted from the first date pattern found in the index name.
+
     """
-    match = re.search(r"\d{4}-\d{2}-\d{2}", index)
-    return datetime_type.strptime(match.group(0), "%Y-%m-%d").date()
+    date_pattern = r'\d{4}-\d{2}-\d{2}'
+    match = re.search(date_pattern, index_name)
+    return datetime_type.strptime(match.group(0), "%Y-%m-%d").date()

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/index_insertion_strategies.py

@@ -9,10 +9,10 @@
 from stac_fastapi.core.utilities import get_bool_env
 from stac_fastapi.sfeos_helpers.database import (
     extract_date,
-    extract_date_from_index,
     index_alias_by_collection_id,
     index_by_collection_id,
     mk_item_id,
+    extract_first_date_from_index,
 )
 from stac_fastapi.sfeos_helpers.mappings import (
     _ES_INDEX_NAME_UNSUPPORTED_CHARS_TABLE,
@@ -54,19 +54,27 @@ def create_datetime_index_sync(
     ) -> str:
         pass
 
-    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 update_index_alias_sync(
-        self, sync_client: Any, collection_id: str, end_date: str
-    ):
-        index = index_alias_by_collection_id(collection_id)
-        sync_client.indices.put_alias(
-            index=index, name=self.alias_by_index_and_end_date(index, end_date)
-        )
+    async def update_index_alias(self, client: Any, end_date: str, old_alias: str):
+        index = ITEMS_INDEX_PREFIX + old_alias
+        new_alias = self.alias_by_index_and_end_date(old_alias, end_date)
+        await client.indices.update_aliases(body={
+            "actions": [
+                {"remove": {"index": index, "alias": old_alias}},
+                {"add": {"index": index, "alias": new_alias}}
+            ]
+        })
+        return new_alias
+
+    def update_index_alias_sync(self, client: Any, end_date: str, old_alias: str):
+        index = ITEMS_INDEX_PREFIX + old_alias
+        new_alias = self.alias_by_index_and_end_date(old_alias, end_date)
+        client.indices.update_aliases(body={
+            "actions": [
+                {"remove": {"index": index, "alias": old_alias}},
+                {"add": {"index": index, "alias": new_alias}}
+            ]
+        })
+        return new_alias
 
     @staticmethod
     def index_by_collection_id_and_date(collection_id: str, start_date: str) -> str:
@@ -247,19 +255,29 @@ async def _get_target_index_base(
             return target_index
 
         all_indexes.sort()
+
+        if (start_date := extract_date(product_datetime)) < (end_date := extract_first_date_from_index(all_indexes[0])):
+            target_index = await self.search_adapter.create_datetime_index(
+                self.client, collection_id, str(start_date)
+            )
+            alias = await self.search_adapter.update_index_alias(
+                self.client, str(end_date - timedelta(days=1)), target_index
+            )
+            await index_selector.refresh_cache()
+            return alias
+
         if target_index != all_indexes[-1]:
             return target_index
 
         if check_size:
-            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 index_size_gb > max_size_gb:
                 end_date = extract_date(product_datetime)
-                if end_date != extract_date_from_index(all_indexes[-1]):
+                if end_date != extract_first_date_from_index(all_indexes[-1]):
                     await self.search_adapter.update_index_alias(
-                        self.client, collection_id, str(end_date)
+                        self.client, str(end_date), target_index
                     )
                     target_index = await self.search_adapter.create_datetime_index(
                         self.client, collection_id, (end_date + timedelta(days=1))
@@ -308,12 +326,12 @@ async def prepare_bulk_actions(
             max_size_gb = float(os.getenv("DATETIME_INDEX_MAX_SIZE_GB", 20))
 
             if index_size_gb > max_size_gb:
-                current_index_end_date = extract_date_from_index(first_item_index)
+                current_index_end_date = extract_first_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.update_index_alias(
-                        self.client, collection_id, str(current_index_end_date)
+                        self.client, str(current_index_end_date), latest_index
                     )
                     next_day_start = current_index_end_date + timedelta(days=1)
                     new_index = await self.search_adapter.create_datetime_index(
@@ -374,6 +392,16 @@ def _get_target_index_base(
             index_selector.refresh_cache()
             return target_index
 
+        if (start_date := extract_date(product_datetime)) < (end_date := extract_first_date_from_index(all_indexes[0])):
+            target_index = self.search_adapter.create_datetime_index_sync(
+                self.sync_client, collection_id, str(start_date)
+            )
+            alias = self.search_adapter.update_index_alias_sync(
+                self.sync_client, str(end_date - timedelta(days=1)), target_index
+            )
+            index_selector.refresh_cache()
+            return alias
+
         all_indexes.sort()
         if target_index != all_indexes[-1]:
             return target_index
@@ -384,9 +412,9 @@ def _get_target_index_base(
 
             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.update_index_alias_sync(
-                        self.sync_client, collection_id, str(end_date)
+                if end_date != extract_first_date_from_index(all_indexes[-1]):
+                    self.search_adapter.update_index_alias(
+                        self.sync_client, str(end_date), target_index
                     )
                     target_index = self.search_adapter.create_datetime_index_sync(
                         self.sync_client, collection_id, (end_date + timedelta(days=1))
@@ -432,12 +460,12 @@ def prepare_bulk_actions(
             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 index_size_gb > max_size_gb:
-                current_index_end_date = extract_date_from_index(first_item_index)
+                current_index_end_date = extract_first_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.update_index_alias_sync(
-                        self.sync_client, collection_id, str(current_index_end_date)
+                        self.sync_client, str(current_index_end_date), latest_index
                     )
                     next_day_start = current_index_end_date + timedelta(days=1)
                     new_index = self.search_adapter.create_datetime_index_sync(

stac_fastapi/tests/api/test_api.py

@@ -4,6 +4,7 @@
 import uuid
 from copy import deepcopy
 from datetime import datetime, timedelta
+from unittest.mock import patch
 
 import pytest
 
@@ -703,7 +704,7 @@ async def test_big_int_eo_search(
 
 
 @pytest.mark.asyncio
-async def test_dodaj_item(app_client, load_test_data, txn_client, ctx):
+async def test_create_item_uses_existing_datetime_index(app_client, load_test_data, txn_client, ctx):
     if not os.getenv("ENABLE_DATETIME_INDEX_FILTERING"):
         pytest.skip()
 
@@ -717,7 +718,7 @@ async def test_dodaj_item(app_client, load_test_data, txn_client, ctx):
 
 
 @pytest.mark.asyncio
-async def test_dodaj_item_2(app_client, load_test_data, txn_client, ctx):
+async def test_create_item_with_different_date_same_index(app_client, load_test_data, txn_client, ctx):
     if not os.getenv("ENABLE_DATETIME_INDEX_FILTERING"):
         pytest.skip()
 
@@ -732,15 +733,35 @@ async def test_dodaj_item_2(app_client, load_test_data, txn_client, ctx):
 
 
 @pytest.mark.asyncio
-async def test_dodaj_item_3(app_client, load_test_data, txn_client, ctx):
+async def test_create_new_index_when_size_limit_exceeded(app_client, load_test_data, txn_client, ctx):
     if not os.getenv("ENABLE_DATETIME_INDEX_FILTERING"):
         pytest.skip()
 
     item = load_test_data("test_item.json")
     item["id"] = str(uuid.uuid4())
-    item["properties"]["datetime"] = "202-02-12T12:30:22Z"
-    created_item = await app_client.post(f"/collections/{item['collection']}/items", json=item)
+    item["properties"]["datetime"] = "2024-02-12T12:30:22Z"
+
+    with patch('stac_fastapi.sfeos_helpers.database.AsyncIndexInserter.get_index_size_in_gb') as mock_get_size:
+        mock_get_size.return_value = 21.0
+        created_item = await app_client.post(f"/collections/{item['collection']}/items", json=item)
+
     assert created_item.status_code == 201
     indices = await txn_client.database.client.indices.get_alias(index="*")
-    assert 'items_test-collection_2020-02-12' in indices.keys()
+    assert 'items_test-collection_2020-02-12' and "items_test-collection_2024-02-13" in indices.keys()
     await app_client.delete(f"/collections/{item['collection']}/items/{item['id']}")
+
+
+@pytest.mark.asyncio
+async def test_create_item_in_past_date_creates_separate_index(app_client, load_test_data, txn_client, ctx):
+    if not os.getenv("ENABLE_DATETIME_INDEX_FILTERING"):
+        pytest.skip()
+
+    item = load_test_data("test_item.json")
+    item["id"] = str(uuid.uuid4())
+    item["properties"]["datetime"] = "2012-02-12T12:30:22Z"
+    created_item = await app_client.post(f"/collections/{item['collection']}/items", json=item)
+    breakpoint()
+    assert created_item.status_code == 201
+    indices = await txn_client.database.client.indices.get_alias(index="*")
+    assert 'items_test-collection_2012-02-12' and 'items_test-collection_2020-02-12' in indices.keys()
+    await app_client.delete(f"/collections/{item['collection']}/items/{item['id']}")