Implement dynamiclly generated queryables

Zaczero · Zaczero · commit 4843f28b5b11 · 2025-04-03T19:39:55.000+02:00
diff --git a/stac_fastapi/core/stac_fastapi/core/core.py b/stac_fastapi/core/stac_fastapi/core/core.py
@@ -1,10 +1,11 @@
 """Core client."""
 
 import logging
+from collections import deque
 from datetime import datetime as datetime_type
 from datetime import timezone
 from enum import Enum
-from typing import Any, Dict, List, Optional, Set, Type, Union
+from typing import Any, Dict, List, Literal, Optional, Set, Type, Union
 from urllib.parse import unquote_plus, urljoin
 
 import attr
@@ -905,11 +906,81 @@ def bulk_item_insert(
         return f"Successfully added {len(processed_items)} Items."
 
 
+_DEFAULT_QUERYABLES = {
+    "id": {
+        "description": "ID",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/2/properties/id",
+    },
+    "collection": {
+        "description": "Collection",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/2/then/properties/collection",
+    },
+    "geometry": {
+        "description": "Geometry",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/1/oneOf/0/properties/geometry",
+    },
+    "datetime": {
+        "description": "Acquisition Timestamp",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/datetime",
+    },
+    "created": {
+        "description": "Creation Timestamp",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/created",
+    },
+    "updated": {
+        "description": "Creation Timestamp",
+        "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/updated",
+    },
+    "cloud_cover": {
+        "description": "Cloud Cover",
+        "$ref": "https://stac-extensions.github.io/eo/v1.0.0/schema.json#/definitions/fields/properties/eo:cloud_cover",
+    },
+    "cloud_shadow_percentage": {
+        "title": "Cloud Shadow Percentage",
+        "description": "Cloud Shadow Percentage",
+        "type": "number",
+        "minimum": 0,
+        "maximum": 100,
+    },
+    "nodata_pixel_percentage": {
+        "title": "No Data Pixel Percentage",
+        "description": "No Data Pixel Percentage",
+        "type": "number",
+        "minimum": 0,
+        "maximum": 100,
+    },
+}
+
+_ES_MAPPING_TYPE_TO_JSON: Dict[
+    str, Literal["string", "number", "boolean", "object", "array", "null"]
+] = {
+    "date": "string",
+    "date_nanos": "string",
+    "keyword": "string",
+    "match_only_text": "string",
+    "text": "string",
+    "wildcard": "string",
+    "byte": "number",
+    "double": "number",
+    "float": "number",
+    "half_float": "number",
+    "long": "number",
+    "scaled_float": "number",
+    "short": "number",
+    "token_count": "number",
+    "unsigned_long": "number",
+    "geo_point": "object",
+    "geo_shape": "object",
+    "nested": "array",
+}
+
+
 @attr.s
 class EsAsyncBaseFiltersClient(AsyncBaseFiltersClient):
     """Defines a pattern for implementing the STAC filter extension."""
 
-    # todo: use the ES _mapping endpoint to dynamically find what fields exist
+    database: BaseDatabaseLogic = attr.ib()
+
     async def get_queryables(
         self, collection_id: Optional[str] = None, **kwargs
     ) -> Dict[str, Any]:
@@ -930,55 +1001,63 @@ async def get_queryables(
         Returns:
             Dict[str, Any]: A dictionary containing the queryables for the given collection.
         """
-        return {
+        queryables: Dict[str, Any] = {
             "$schema": "https://json-schema.org/draft/2019-09/schema",
             "$id": "https://stac-api.example.com/queryables",
             "type": "object",
-            "title": "Queryables for Example STAC API",
-            "description": "Queryable names for the example STAC API Item Search filter.",
-            "properties": {
-                "id": {
-                    "description": "ID",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/2/properties/id",
-                },
-                "collection": {
-                    "description": "Collection",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/2/then/properties/collection",
-                },
-                "geometry": {
-                    "description": "Geometry",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/definitions/core/allOf/1/oneOf/0/properties/geometry",
-                },
-                "datetime": {
-                    "description": "Acquisition Timestamp",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/datetime",
-                },
-                "created": {
-                    "description": "Creation Timestamp",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/created",
-                },
-                "updated": {
-                    "description": "Creation Timestamp",
-                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/datetime.json#/properties/updated",
-                },
-                "cloud_cover": {
-                    "description": "Cloud Cover",
-                    "$ref": "https://stac-extensions.github.io/eo/v1.0.0/schema.json#/definitions/fields/properties/eo:cloud_cover",
-                },
-                "cloud_shadow_percentage": {
-                    "description": "Cloud Shadow Percentage",
-                    "title": "Cloud Shadow Percentage",
-                    "type": "number",
-                    "minimum": 0,
-                    "maximum": 100,
-                },
-                "nodata_pixel_percentage": {
-                    "description": "No Data Pixel Percentage",
-                    "title": "No Data Pixel Percentage",
-                    "type": "number",
-                    "minimum": 0,
-                    "maximum": 100,
-                },
-            },
+            "title": "Queryables for STAC API",
+            "description": "Queryable names for the STAC API Item Search filter.",
+            "properties": _DEFAULT_QUERYABLES,
             "additionalProperties": True,
         }
+        if not collection_id:
+            return queryables
+
+        properties = {}
+        queryables.update(
+            {
+                "properties": properties,
+                "additionalProperties": False,
+            }
+        )
+
+        mapping_data = await self.database.get_items_mapping(collection_id)
+        mapping_properties = next(iter(mapping_data.values()))["mappings"]["properties"]
+        stack = deque(mapping_properties.items())
+
+        while stack:
+            field_name, field_def = stack.popleft()
+
+            # Iterate over nested fields
+            field_properties = field_def.get("properties")
+            if field_properties:
+                # Fields in Item Properties should be exposed with their un-prefixed names,
+                # and not require expressions to prefix them with properties,
+                # e.g., eo:cloud_cover instead of properties.eo:cloud_cover.
+                if field_name == "properties":
+                    stack.extend(field_properties.items())
+                else:
+                    stack.extend(
+                        (f"{field_name}.{k}", v) for k, v in field_properties.items()
+                    )
+
+            # Skip non-indexed or disabled fields
+            field_type = field_def.get("type")
+            if not field_type or not field_def.get("enabled", True):
+                continue
+
+            # Generate field properties
+            field_result = _DEFAULT_QUERYABLES.get(field_name, {})
+            properties[field_name] = field_result
+
+            field_name_human = field_name.replace("_", " ").title()
+            field_result.setdefault("title", field_name_human)
+            field_result.setdefault("description", field_name_human)
+
+            field_type_json = _ES_MAPPING_TYPE_TO_JSON.get(field_type, field_type)
+            field_result.setdefault("type", field_type_json)
+
+            if field_type in {"date", "date_nanos"}:
+                field_result.setdefault("format", "date-time")
+
+        return queryables
diff --git a/stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py b/stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py
@@ -39,13 +39,15 @@
 settings = ElasticsearchSettings()
 session = Session.create_from_settings(settings)
 
-filter_extension = FilterExtension(client=EsAsyncBaseFiltersClient())
+database_logic = DatabaseLogic()
+
+filter_extension = FilterExtension(
+    client=EsAsyncBaseFiltersClient(database=database_logic)
+)
 filter_extension.conformance_classes.append(
     "http://www.opengis.net/spec/cql2/1.0/conf/advanced-comparison-operators"
 )
 
-database_logic = DatabaseLogic()
-
 aggregation_extension = AggregationExtension(
     client=EsAsyncAggregationClient(
         database=database_logic, session=session, settings=settings
diff --git a/stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py b/stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py
@@ -725,6 +725,24 @@ async def delete_item(
                 f"Item {item_id} in collection {collection_id} not found"
             )
 
+    async def get_items_mapping(self, collection_id: str) -> Dict[str, Any]:
+        """Get the mapping for the specified collection's items index.
+
+        Args:
+            collection_id (str): The ID of the collection to get items mapping for.
+
+        Returns:
+            Dict[str, Any]: The mapping information.
+        """
+        index_name = index_alias_by_collection_id(collection_id)
+        try:
+            mapping = await self.client.indices.get_mapping(
+                index=index_name, allow_no_indices=False
+            )
+            return mapping.body
+        except exceptions.NotFoundError:
+            raise NotFoundError(f"Mapping for index {index_name} not found")
+
     async def create_collection(self, collection: Collection, refresh: bool = False):
         """Create a single collection in the database.
 
diff --git a/stac_fastapi/opensearch/stac_fastapi/opensearch/app.py b/stac_fastapi/opensearch/stac_fastapi/opensearch/app.py
@@ -39,13 +39,15 @@
 settings = OpensearchSettings()
 session = Session.create_from_settings(settings)
 
-filter_extension = FilterExtension(client=EsAsyncBaseFiltersClient())
+database_logic = DatabaseLogic()
+
+filter_extension = FilterExtension(
+    client=EsAsyncBaseFiltersClient(database=database_logic)
+)
 filter_extension.conformance_classes.append(
     "http://www.opengis.net/spec/cql2/1.0/conf/advanced-comparison-operators"
 )
 
-database_logic = DatabaseLogic()
-
 aggregation_extension = AggregationExtension(
     client=EsAsyncAggregationClient(
         database=database_logic, session=session, settings=settings
diff --git a/stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py b/stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py
@@ -757,6 +757,24 @@ async def delete_item(
                 f"Item {item_id} in collection {collection_id} not found"
             )
 
+    async def get_items_mapping(self, collection_id: str) -> Dict[str, Any]:
+        """Get the mapping for the specified collection's items index.
+
+        Args:
+            collection_id (str): The ID of the collection to get items mapping for.
+
+        Returns:
+            Dict[str, Any]: The mapping information.
+        """
+        index_name = index_alias_by_collection_id(collection_id)
+        try:
+            mapping = await self.client.indices.get_mapping(
+                index=index_name, params={"allow_no_indices": False}
+            )
+            return mapping
+        except exceptions.NotFoundError:
+            raise NotFoundError(f"Mapping for index {index_name} not found")
+
     async def create_collection(self, collection: Collection, refresh: bool = False):
         """Create a single collection in the database.
 
