move utility functions

Author: jonhealy1

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/config.py

@@ -10,7 +10,8 @@
 
 from elasticsearch import Elasticsearch  # type: ignore[attr-defined]
 from stac_fastapi.core.base_settings import ApiBaseSettings
-from stac_fastapi.sfeos_helpers.utilities import get_bool_env, validate_refresh
+from stac_fastapi.core.utilities import get_bool_env
+from stac_fastapi.sfeos_helpers.database import validate_refresh
 from stac_fastapi.types.config import ApiSettings
 
 

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -27,7 +27,13 @@
     create_index_templates_shared,
     delete_item_index_shared,
     get_queryables_mapping_shared,
+    index_alias_by_collection_id,
+    index_by_collection_id,
+    indices,
+    mk_actions,
+    mk_item_id,
     populate_sort_shared,
+    validate_refresh,
 )
 from stac_fastapi.sfeos_helpers.mappings import (
     AGGREGATION_MAPPING,
@@ -37,14 +43,6 @@
     ITEMS_INDEX_PREFIX,
     Geometry,
 )
-from stac_fastapi.sfeos_helpers.utilities import (
-    index_alias_by_collection_id,
-    index_by_collection_id,
-    indices,
-    mk_actions,
-    mk_item_id,
-    validate_refresh,
-)
 from stac_fastapi.types.errors import ConflictError, NotFoundError
 from stac_fastapi.types.stac import Collection, Item
 

stac_fastapi/opensearch/stac_fastapi/opensearch/config.py

@@ -8,7 +8,8 @@
 from opensearchpy import AsyncOpenSearch, OpenSearch
 
 from stac_fastapi.core.base_settings import ApiBaseSettings
-from stac_fastapi.sfeos_helpers.utilities import get_bool_env, validate_refresh
+from stac_fastapi.core.utilities import get_bool_env
+from stac_fastapi.sfeos_helpers.database import validate_refresh
 from stac_fastapi.types.config import ApiSettings
 
 

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -27,7 +27,13 @@
     create_index_templates_shared,
     delete_item_index_shared,
     get_queryables_mapping_shared,
+    index_alias_by_collection_id,
+    index_by_collection_id,
+    indices,
+    mk_actions,
+    mk_item_id,
     populate_sort_shared,
+    validate_refresh,
 )
 from stac_fastapi.sfeos_helpers.mappings import (
     AGGREGATION_MAPPING,
@@ -40,14 +46,6 @@
     ITEMS_INDEX_PREFIX,
     Geometry,
 )
-from stac_fastapi.sfeos_helpers.utilities import (
-    index_alias_by_collection_id,
-    index_by_collection_id,
-    indices,
-    mk_actions,
-    mk_item_id,
-    validate_refresh,
-)
 from stac_fastapi.types.errors import ConflictError, NotFoundError
 from stac_fastapi.types.stac import Collection, Item
 

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/README.md

@@ -6,11 +6,14 @@ between the two implementations.
 
 ## Package Structure
 
-The database package is organized into three main modules:
+The database package is organized into five main modules:
 
 - **index.py**: Contains functions for managing indices
   - [create_index_templates_shared](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database_logic_helpers.py:15:0-48:33): Creates index templates for Collections and Items
   - [delete_item_index_shared](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database_logic_helpers.py:128:0-153:30): Deletes an item index for a collection
+  - [index_by_collection_id](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:86:0-100:5): Translates a collection ID into an index name
+  - [index_alias_by_collection_id](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:103:0-115:5): Translates a collection ID into an index alias
+  - [indices](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:118:0-132:5): Gets a comma-separated string of index names
 
 - **query.py**: Contains functions for building and manipulating queries
   - [apply_free_text_filter_shared](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database_logic_helpers.py:51:0-74:16): Applies a free text filter to a search
@@ -20,16 +23,39 @@ The database package is organized into three main modules:
 - **mapping.py**: Contains functions for working with mappings
   - [get_queryables_mapping_shared](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database_logic_helpers.py:156:0-185:27): Retrieves mapping of Queryables for search
 
+- **document.py**: Contains functions for working with documents
+  - [mk_item_id](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:140:0-150:5): Creates a document ID for an Item
+  - [mk_actions](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:153:0-175:5): Creates bulk actions for indexing items
+
+- **utils.py**: Contains utility functions for database operations
+  - [validate_refresh](cci:1://file:///home/computer/Code/stac-fastapi-elasticsearch-opensearch/stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/utilities.py:41:0-78:5): Validates the refresh parameter value
+
 ## Usage
 
 Import the necessary components from the database package:
 
 ```python
 from stac_fastapi.sfeos_helpers.database import (
+    # Index operations
     create_index_templates_shared,
     delete_item_index_shared,
+    index_alias_by_collection_id,
+    index_by_collection_id,
+    indices,
+    
+    # Query operations
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     populate_sort_shared,
+    
+    # Mapping operations
     get_queryables_mapping_shared,
-)
+    
+    # Document operations
+    mk_item_id,
+    mk_actions,
+    
+    # Utility functions
+    validate_refresh,
+)
+```

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

@@ -6,11 +6,15 @@
 1. Index management functions for creating and deleting indices
 2. Query building functions for constructing search queries
 3. Mapping functions for working with Elasticsearch/OpenSearch mappings
+4. Document operations for working with documents
+5. Utility functions for database operations
 
 The database package is organized as follows:
 - index.py: Index management functions
 - query.py: Query building functions
 - mapping.py: Mapping functions
+- document.py: Document operations
+- utils.py: Utility functions
 
 When adding new functionality to this package, consider:
 1. Will this code be used by both Elasticsearch and OpenSearch implementations?
@@ -24,19 +28,38 @@
 """
 
 # Re-export all functions for backward compatibility
-from .index import create_index_templates_shared, delete_item_index_shared
+from .document import mk_actions, mk_item_id
+from .index import (
+    create_index_templates_shared,
+    delete_item_index_shared,
+    index_alias_by_collection_id,
+    index_by_collection_id,
+    indices,
+)
 from .mapping import get_queryables_mapping_shared
 from .query import (
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     populate_sort_shared,
 )
+from .utils import validate_refresh
 
 __all__ = [
+    # Index operations
     "create_index_templates_shared",
     "delete_item_index_shared",
+    "index_alias_by_collection_id",
+    "index_by_collection_id",
+    "indices",
+    # Query operations
     "apply_free_text_filter_shared",
     "apply_intersects_filter_shared",
     "populate_sort_shared",
+    # Mapping operations
     "get_queryables_mapping_shared",
+    # Document operations
+    "mk_item_id",
+    "mk_actions",
+    # Utility functions
+    "validate_refresh",
 ]

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

@@ -0,0 +1,48 @@
+"""Document operations for Elasticsearch/OpenSearch.
+
+This module provides functions for working with documents in Elasticsearch/OpenSearch,
+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.
+
+    Args:
+        item_id (str): The id of the Item.
+        collection_id (str): The id of the Collection that the Item belongs to.
+
+    Returns:
+        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/database/index.py

@@ -3,16 +3,67 @@
 This module provides functions for creating and managing indices in Elasticsearch/OpenSearch.
 """
 
-from typing import Any
+from functools import lru_cache
+from typing import Any, List, Optional
 
 from stac_fastapi.sfeos_helpers.mappings import (
+    _ES_INDEX_NAME_UNSUPPORTED_CHARS_TABLE,
     COLLECTIONS_INDEX,
     ES_COLLECTIONS_MAPPINGS,
     ES_ITEMS_MAPPINGS,
     ES_ITEMS_SETTINGS,
+    ITEM_INDICES,
     ITEMS_INDEX_PREFIX,
 )
-from stac_fastapi.sfeos_helpers.utilities import index_alias_by_collection_id
+
+
+@lru_cache(256)
+def index_by_collection_id(collection_id: str) -> str:
+    """
+    Translate a collection id into an Elasticsearch index name.
+
+    Args:
+        collection_id (str): The collection id to translate into an index name.
+
+    Returns:
+        str: The index name derived from the collection id.
+    """
+    cleaned = collection_id.translate(_ES_INDEX_NAME_UNSUPPORTED_CHARS_TABLE)
+    return (
+        f"{ITEMS_INDEX_PREFIX}{cleaned.lower()}_{collection_id.encode('utf-8').hex()}"
+    )
+
+
+@lru_cache(256)
+def index_alias_by_collection_id(collection_id: str) -> str:
+    """
+    Translate a collection id into an Elasticsearch index alias.
+
+    Args:
+        collection_id (str): The collection id to translate into an index alias.
+
+    Returns:
+        str: The index alias derived from the collection id.
+    """
+    cleaned = collection_id.translate(_ES_INDEX_NAME_UNSUPPORTED_CHARS_TABLE)
+    return f"{ITEMS_INDEX_PREFIX}{cleaned}"
+
+
+def indices(collection_ids: Optional[List[str]]) -> str:
+    """
+    Get a comma-separated string of index names for a given list of collection ids.
+
+    Args:
+        collection_ids: A list of collection ids.
+
+    Returns:
+        A string of comma-separated index names. If `collection_ids` is empty, returns the default indices.
+    """
+    return (
+        ",".join(map(index_alias_by_collection_id, collection_ids))
+        if collection_ids
+        else ITEM_INDICES
+    )
 
 
 async def create_index_templates_shared(settings: Any) -> None:

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/utils.py

@@ -0,0 +1,50 @@
+"""Utility functions for database operations in Elasticsearch/OpenSearch.
+
+This module provides utility functions for working with database operations
+in Elasticsearch/OpenSearch, such as parameter validation.
+"""
+
+import logging
+from typing import Union
+
+from stac_fastapi.core.utilities import get_bool_env
+
+
+def validate_refresh(value: Union[str, bool]) -> str:
+    """
+    Validate the `refresh` parameter value.
+
+    Args:
+        value (Union[str, bool]): The `refresh` parameter value, which can be a string or a boolean.
+
+    Returns:
+        str: The validated value of the `refresh` parameter, which can be "true", "false", or "wait_for".
+    """
+    logger = logging.getLogger(__name__)
+
+    # Handle boolean-like values using get_bool_env
+    if isinstance(value, bool) or value in {
+        "true",
+        "false",
+        "1",
+        "0",
+        "yes",
+        "no",
+        "y",
+        "n",
+    }:
+        is_true = get_bool_env("DATABASE_REFRESH", default=value)
+        return "true" if is_true else "false"
+
+    # Normalize to lowercase for case-insensitivity
+    value = value.lower()
+
+    # Handle "wait_for" explicitly
+    if value == "wait_for":
+        return "wait_for"
+
+    # Log a warning for invalid values and default to "false"
+    logger.warning(
+        f"Invalid value for `refresh`: '{value}'. Expected 'true', 'false', or 'wait_for'. Defaulting to 'false'."
+    )
+    return "false"