move return_date

Author: jonhealy1

stac_fastapi/core/stac_fastapi/core/core.py

@@ -21,7 +21,7 @@
 
 from stac_fastapi.core.base_database_logic import BaseDatabaseLogic
 from stac_fastapi.core.base_settings import ApiBaseSettings
-from stac_fastapi.core.datetime_utils import format_datetime_range, return_date
+from stac_fastapi.core.datetime_utils import format_datetime_range
 from stac_fastapi.core.models.links import PagingLinks
 from stac_fastapi.core.serializers import CollectionSerializer, ItemSerializer
 from stac_fastapi.core.session import Session
@@ -316,9 +316,8 @@ async def item_collection(
         )
 
         if datetime:
-            datetime_search = return_date(datetime)
             search = self.database.apply_datetime_filter(
-                search=search, datetime_search=datetime_search
+                search=search, interval=datetime
             )
 
         if bbox:
@@ -493,9 +492,8 @@ async def post_search(
             )
 
         if search_request.datetime:
-            datetime_search = return_date(search_request.datetime)
             search = self.database.apply_datetime_filter(
-                search=search, datetime_search=datetime_search
+                search=search, interval=search_request.datetime
             )
 
         if search_request.bbox:

stac_fastapi/core/stac_fastapi/core/datetime_utils.py

@@ -1,60 +1,7 @@
 """Utility functions to handle datetime parsing."""
-from datetime import datetime
-from datetime import datetime as datetime_type
-from datetime import timezone
-from typing import Dict, Optional, Union
+from datetime import datetime, timezone
 
-from stac_fastapi.types.rfc3339 import DateTimeType, rfc3339_str_to_datetime
-
-
-def return_date(
-    interval: Optional[Union[DateTimeType, str]]
-) -> Dict[str, Optional[str]]:
-    """
-    Convert a date interval.
-
-    (which may be a datetime, a tuple of one or two datetimes a string
-    representing a datetime or range, or None) into a dictionary for filtering
-    search results with Elasticsearch.
-
-    This function ensures the output dictionary contains 'gte' and 'lte' keys,
-    even if they are set to None, to prevent KeyError in the consuming logic.
-
-    Args:
-        interval (Optional[Union[DateTimeType, str]]): The date interval, which might be a single datetime,
-            a tuple with one or two datetimes, a string, or None.
-
-    Returns:
-        dict: A dictionary representing the date interval for use in filtering search results,
-            always containing 'gte' and 'lte' keys.
-    """
-    result: Dict[str, Optional[str]] = {"gte": None, "lte": None}
-
-    if interval is None:
-        return result
-
-    if isinstance(interval, str):
-        if "/" in interval:
-            parts = interval.split("/")
-            result["gte"] = parts[0] if parts[0] != ".." else None
-            result["lte"] = parts[1] if len(parts) > 1 and parts[1] != ".." else None
-        else:
-            converted_time = interval if interval != ".." else None
-            result["gte"] = result["lte"] = converted_time
-        return result
-
-    if isinstance(interval, datetime_type):
-        datetime_iso = interval.isoformat()
-        result["gte"] = result["lte"] = datetime_iso
-    elif isinstance(interval, tuple):
-        start, end = interval
-        # Ensure datetimes are converted to UTC and formatted with 'Z'
-        if start:
-            result["gte"] = start.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
-        if end:
-            result["lte"] = end.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
-
-    return result
+from stac_fastapi.types.rfc3339 import rfc3339_str_to_datetime
 
 
 def format_datetime_range(date_str: str) -> str:

stac_fastapi/core/stac_fastapi/core/extensions/aggregation.py

@@ -1,7 +1,6 @@
 """Request model for the Aggregation extension."""
 
 from datetime import datetime
-from datetime import datetime as datetime_type
 from typing import Dict, List, Literal, Optional, Union
 from urllib.parse import unquote_plus, urljoin
 
@@ -210,58 +209,6 @@ def extract_date_histogram_interval(self, value: Optional[str]) -> str:
         else:
             return self.DEFAULT_DATETIME_INTERVAL
 
-    @staticmethod
-    def _return_date(
-        interval: Optional[Union[DateTimeType, str]]
-    ) -> Dict[str, Optional[str]]:
-        """
-        Convert a date interval.
-
-        (which may be a datetime, a tuple of one or two datetimes a string
-        representing a datetime or range, or None) into a dictionary for filtering
-        search results with Elasticsearch.
-
-        This function ensures the output dictionary contains 'gte' and 'lte' keys,
-        even if they are set to None, to prevent KeyError in the consuming logic.
-
-        Args:
-            interval (Optional[Union[DateTimeType, str]]): The date interval, which might be a single datetime,
-                a tuple with one or two datetimes, a string, or None.
-
-        Returns:
-            dict: A dictionary representing the date interval for use in filtering search results,
-                always containing 'gte' and 'lte' keys.
-        """
-        result: Dict[str, Optional[str]] = {"gte": None, "lte": None}
-
-        if interval is None:
-            return result
-
-        if isinstance(interval, str):
-            if "/" in interval:
-                parts = interval.split("/")
-                result["gte"] = parts[0] if parts[0] != ".." else None
-                result["lte"] = (
-                    parts[1] if len(parts) > 1 and parts[1] != ".." else None
-                )
-            else:
-                converted_time = interval if interval != ".." else None
-                result["gte"] = result["lte"] = converted_time
-            return result
-
-        if isinstance(interval, datetime_type):
-            datetime_iso = interval.isoformat()
-            result["gte"] = result["lte"] = datetime_iso
-        elif isinstance(interval, tuple):
-            start, end = interval
-            # Ensure datetimes are converted to UTC and formatted with 'Z'
-            if start:
-                result["gte"] = start.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
-            if end:
-                result["lte"] = end.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
-
-        return result
-
     def frequency_agg(self, es_aggs, name, data_type):
         """Format an aggregation for a frequency distribution aggregation."""
         buckets = []
@@ -418,9 +365,8 @@ async def aggregate(
             )
 
         if aggregate_request.datetime:
-            datetime_search = self._return_date(aggregate_request.datetime)
             search = self.database.apply_datetime_filter(
-                search=search, datetime_search=datetime_search
+                search=search, interval=aggregate_request.datetime
             )
 
         if aggregate_request.bbox:

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -5,7 +5,7 @@
 import logging
 from base64 import urlsafe_b64decode, urlsafe_b64encode
 from copy import deepcopy
-from typing import Any, Dict, Iterable, List, Optional, Tuple, Type
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
 
 import attr
 import elasticsearch.helpers as helpers
@@ -33,6 +33,7 @@
     mk_actions,
     mk_item_id,
     populate_sort_shared,
+    return_date,
     validate_refresh,
 )
 from stac_fastapi.sfeos_helpers.mappings import (
@@ -44,6 +45,7 @@
     Geometry,
 )
 from stac_fastapi.types.errors import ConflictError, NotFoundError
+from stac_fastapi.types.rfc3339 import DateTimeType
 from stac_fastapi.types.stac import Collection, Item
 
 logger = logging.getLogger(__name__)
@@ -241,17 +243,20 @@ def apply_collections_filter(search: Search, collection_ids: List[str]):
         return search.filter("terms", collection=collection_ids)
 
     @staticmethod
-    def apply_datetime_filter(search: Search, datetime_search: dict):
+    def apply_datetime_filter(
+        search: Search, interval: Optional[Union[DateTimeType, str]]
+    ):
         """Apply a filter to search on datetime, start_datetime, and end_datetime fields.
 
         Args:
             search (Search): The search object to filter.
-            datetime_search (dict): The datetime filter criteria.
+            interval: Optional[Union[DateTimeType, str]]
 
         Returns:
             Search: The filtered search object.
         """
         should = []
+        datetime_search = return_date(interval)
 
         # If the request is a single datetime return
         # items with datetimes equal to the requested datetime OR

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -5,7 +5,7 @@
 import logging
 from base64 import urlsafe_b64decode, urlsafe_b64encode
 from copy import deepcopy
-from typing import Any, Dict, Iterable, List, Optional, Tuple, Type
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
 
 import attr
 from opensearchpy import exceptions, helpers
@@ -33,6 +33,7 @@
     mk_actions,
     mk_item_id,
     populate_sort_shared,
+    return_date,
     validate_refresh,
 )
 from stac_fastapi.sfeos_helpers.mappings import (
@@ -47,6 +48,7 @@
     Geometry,
 )
 from stac_fastapi.types.errors import ConflictError, NotFoundError
+from stac_fastapi.types.rfc3339 import DateTimeType
 from stac_fastapi.types.stac import Collection, Item
 
 logger = logging.getLogger(__name__)
@@ -278,17 +280,20 @@ def apply_free_text_filter(search: Search, free_text_queries: Optional[List[str]
         )
 
     @staticmethod
-    def apply_datetime_filter(search: Search, datetime_search):
+    def apply_datetime_filter(
+        search: Search, interval: Optional[Union[DateTimeType, str]]
+    ):
         """Apply a filter to search based on datetime field, start_datetime, and end_datetime fields.
 
         Args:
             search (Search): The search object to filter.
-            datetime_search (dict): The datetime filter criteria.
+            interval: Optional[Union[DateTimeType, str]]
 
         Returns:
             Search: The filtered search object.
         """
         should = []
+        datetime_search = return_date(interval)
 
         # If the request is a single datetime return
         # items with datetimes equal to the requested datetime OR

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

@@ -8,13 +8,15 @@
 3. Mapping functions for working with Elasticsearch/OpenSearch mappings
 4. Document operations for working with documents
 5. Utility functions for database operations
+6. Datetime utilities for query formatting
 
 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
+- datetime.py: Datetime utilities for query formatting
 
 When adding new functionality to this package, consider:
 1. Will this code be used by both Elasticsearch and OpenSearch implementations?
@@ -28,6 +30,7 @@
 """
 
 # Re-export all functions for backward compatibility
+from .datetime import return_date
 from .document import mk_actions, mk_item_id
 from .index import (
     create_index_templates_shared,
@@ -42,7 +45,7 @@
     apply_intersects_filter_shared,
     populate_sort_shared,
 )
-from .utils import validate_refresh
+from .utils import get_bool_env, validate_refresh
 
 __all__ = [
     # Index operations
@@ -62,4 +65,7 @@
     "mk_actions",
     # Utility functions
     "validate_refresh",
+    "get_bool_env",
+    # Datetime utilities
+    "return_date",
 ]

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

@@ -0,0 +1,60 @@
+"""Elasticsearch/OpenSearch-specific datetime utilities.
+
+This module provides datetime utility functions specifically designed for
+Elasticsearch and OpenSearch query formatting.
+"""
+
+from datetime import datetime as datetime_type
+from typing import Dict, Optional, Union
+
+from stac_fastapi.types.rfc3339 import DateTimeType
+
+
+def return_date(
+    interval: Optional[Union[DateTimeType, str]]
+) -> Dict[str, Optional[str]]:
+    """
+    Convert a date interval to an Elasticsearch/OpenSearch query format.
+
+    This function converts a date interval (which may be a datetime, a tuple of one or two datetimes,
+    a string representing a datetime or range, or None) into a dictionary for filtering
+    search results with Elasticsearch/OpenSearch.
+
+    This function ensures the output dictionary contains 'gte' and 'lte' keys,
+    even if they are set to None, to prevent KeyError in the consuming logic.
+
+    Args:
+        interval (Optional[Union[DateTimeType, str]]): The date interval, which might be a single datetime,
+            a tuple with one or two datetimes, a string, or None.
+
+    Returns:
+        dict: A dictionary representing the date interval for use in filtering search results,
+            always containing 'gte' and 'lte' keys.
+    """
+    result: Dict[str, Optional[str]] = {"gte": None, "lte": None}
+
+    if interval is None:
+        return result
+
+    if isinstance(interval, str):
+        if "/" in interval:
+            parts = interval.split("/")
+            result["gte"] = parts[0] if parts[0] != ".." else None
+            result["lte"] = parts[1] if len(parts) > 1 and parts[1] != ".." else None
+        else:
+            converted_time = interval if interval != ".." else None
+            result["gte"] = result["lte"] = converted_time
+        return result
+
+    if isinstance(interval, datetime_type):
+        datetime_iso = interval.isoformat()
+        result["gte"] = result["lte"] = datetime_iso
+    elif isinstance(interval, tuple):
+        start, end = interval
+        # Ensure datetimes are converted to UTC and formatted with 'Z'
+        if start:
+            result["gte"] = start.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+        if end:
+            result["lte"] = end.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+    return result