add use_datetime to apply_datetime_filter

Author: Yuri Zmytrakov

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -1,5 +1,5 @@
 """Database logic."""
-
+from stac_fastapi.core.utilities import get_bool_env
 import asyncio
 import logging
 from base64 import urlsafe_b64decode, urlsafe_b64encode
@@ -301,41 +301,22 @@ def apply_datetime_filter(
         if not datetime_search:
             return search, datetime_search
 
-        if "eq" in datetime_search:
-            # For exact matches, include:
-            # 1. Items with matching exact datetime
-            # 2. Items with datetime:null where the time falls within their range
-            should = [
-                Q(
+        # USE_DATETIME env var
+        # True: Only search by datetime and ignore start/end datetime
+        # False: Search by datetime, if null search by start/end datetime
+        USE_DATETIME = get_bool_env("USE_DATETIME", default=False)
+
+        if USE_DATETIME:
+            if "eq" in datetime_search:
+                filter_query = Q(
                     "bool",
                     filter=[
                         Q("exists", field="properties.datetime"),
                         Q("term", **{"properties__datetime": datetime_search["eq"]}),
                     ],
-                ),
-                Q(
-                    "bool",
-                    must_not=[Q("exists", field="properties.datetime")],
-                    filter=[
-                        Q("exists", field="properties.start_datetime"),
-                        Q("exists", field="properties.end_datetime"),
-                        Q(
-                            "range",
-                            properties__start_datetime={"lte": datetime_search["eq"]},
-                        ),
-                        Q(
-                            "range",
-                            properties__end_datetime={"gte": datetime_search["eq"]},
-                        ),
-                    ],
-                ),
-            ]
-        else:
-            # For date ranges, include:
-            # 1. Items with datetime in the range
-            # 2. Items with datetime:null that overlap the search range
-            should = [
-                Q(
+                )
+            else:
+                filter_query = Q(
                     "bool",
                     filter=[
                         Q("exists", field="properties.datetime"),
@@ -347,29 +328,75 @@ def apply_datetime_filter(
                             },
                         ),
                     ],
-                ),
-                Q(
-                    "bool",
-                    must_not=[Q("exists", field="properties.datetime")],
-                    filter=[
-                        Q("exists", field="properties.start_datetime"),
-                        Q("exists", field="properties.end_datetime"),
-                        Q(
-                            "range",
-                            properties__start_datetime={"lte": datetime_search["lte"]},
-                        ),
-                        Q(
-                            "range",
-                            properties__end_datetime={"gte": datetime_search["gte"]},
-                        ),
-                    ],
-                ),
-            ]
-
-        return (
-            search.query(Q("bool", should=should, minimum_should_match=1)),
-            datetime_search,
-        )
+                )
+            return search.query(filter_query), datetime_search
+        else:
+            if "eq" in datetime_search:
+                should = [
+                    Q(
+                        "bool",
+                        filter=[
+                            Q("exists", field="properties.datetime"),
+                            Q("term", **{"properties__datetime": datetime_search["eq"]}),
+                        ],
+                    ),
+                    Q(
+                        "bool",
+                        must_not=[Q("exists", field="properties.datetime")],
+                        filter=[
+                            Q("exists", field="properties.start_datetime"),
+                            Q("exists", field="properties.end_datetime"),
+                            Q(
+                                "range",
+                                properties__start_datetime={"lte": datetime_search["eq"]},
+                            ),
+                            Q(
+                                "range",
+                                properties__end_datetime={"gte": datetime_search["eq"]},
+                            ),
+                        ],
+                    ),
+                ]
+            else:
+                # For date ranges, include:
+                # 1. Items with datetime in the range
+                # 2. Items with datetime:null that overlap the search range
+                should = [
+                    Q(
+                        "bool",
+                        filter=[
+                            Q("exists", field="properties.datetime"),
+                            Q(
+                                "range",
+                                properties__datetime={
+                                    "gte": datetime_search["gte"],
+                                    "lte": datetime_search["lte"],
+                                },
+                            ),
+                        ],
+                    ),
+                    Q(
+                        "bool",
+                        must_not=[Q("exists", field="properties.datetime")],
+                        filter=[
+                            Q("exists", field="properties.start_datetime"),
+                            Q("exists", field="properties.end_datetime"),
+                            Q(
+                                "range",
+                                properties__start_datetime={"lte": datetime_search["lte"]},
+                            ),
+                            Q(
+                                "range",
+                                properties__end_datetime={"gte": datetime_search["gte"]},
+                            ),
+                        ],
+                    ),
+                ]
+
+            return (
+                search.query(Q("bool", should=should, minimum_should_match=1)),
+                datetime_search,
+            )
 
     @staticmethod
     def apply_bbox_filter(search: Search, bbox: List):