Merge branch 'main' of github.com:stac-utils/stac-fastapi-elasticsearch into patch_endpoints

Author: rhysrevans3

CHANGELOG.md

@@ -8,6 +8,19 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+
+- Added support for enum queryables [#390](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/390)
+
+### Changed
+
+- Optimize data_loader.py script [#395](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/395)
+- Refactored test configuration to use shared app config pattern [#399](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/399)
+
+### Removed
+
+- Removed `requests` dev dependency [#395](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/395)
+
 ## [v5.0.0a1] - 2025-05-30
 
 ### Changed

Makefile

@@ -3,13 +3,11 @@ APP_HOST ?= 0.0.0.0
 EXTERNAL_APP_PORT ?= 8080
 
 ES_APP_PORT ?= 8080
+OS_APP_PORT ?= 8082
+
 ES_HOST ?= docker.for.mac.localhost
 ES_PORT ?= 9200
 
-OS_APP_PORT ?= 8082
-OS_HOST ?= docker.for.mac.localhost
-OS_PORT ?= 9202
-
 run_es = docker compose \
 	run \
 	-p ${EXTERNAL_APP_PORT}:${ES_APP_PORT} \

data_loader.py

@@ -1,106 +1,105 @@
 """Data Loader CLI STAC_API Ingestion Tool."""
-import json
+
 import os
+from typing import Any
 
 import click
-import requests
+import orjson
+from httpx import Client
 
 
-def load_data(data_dir, filename):
+def load_data(filepath: str) -> dict[str, Any]:
     """Load json data from a file within the specified data directory."""
-    filepath = os.path.join(data_dir, filename)
-    if not os.path.exists(filepath):
+    try:
+        with open(filepath, "rb") as file:
+            return orjson.loads(file.read())
+    except FileNotFoundError as e:
         click.secho(f"File not found: {filepath}", fg="red", err=True)
-        raise click.Abort()
-    with open(filepath) as file:
-        return json.load(file)
+        raise click.Abort() from e
 
 
-def load_collection(base_url, collection_id, data_dir):
+def load_collection(client: Client, collection_id: str, data_dir: str) -> None:
     """Load a STAC collection into the database."""
-    collection = load_data(data_dir, "collection.json")
+    collection = load_data(os.path.join(data_dir, "collection.json"))
     collection["id"] = collection_id
-    try:
-        resp = requests.post(f"{base_url}/collections", json=collection)
-        if resp.status_code == 200 or resp.status_code == 201:
-            click.echo(f"Status code: {resp.status_code}")
-            click.echo(f"Added collection: {collection['id']}")
-        elif resp.status_code == 409:
-            click.echo(f"Status code: {resp.status_code}")
-            click.echo(f"Collection: {collection['id']} already exists")
-        else:
-            click.echo(f"Status code: {resp.status_code}")
-            click.echo(
-                f"Error writing {collection['id']} collection. Message: {resp.text}"
-            )
-    except requests.ConnectionError:
-        click.secho("Failed to connect", fg="red", err=True)
+    resp = client.post("/collections", json=collection)
+    if resp.status_code == 200 or resp.status_code == 201:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo(f"Added collection: {collection['id']}")
+    elif resp.status_code == 409:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo(f"Collection: {collection['id']} already exists")
+    else:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo(f"Error writing {collection['id']} collection. Message: {resp.text}")
 
 
-def load_items(base_url, collection_id, use_bulk, data_dir):
+def load_items(
+    client: Client, collection_id: str, use_bulk: bool, data_dir: str
+) -> None:
     """Load STAC items into the database based on the method selected."""
-    # Attempt to dynamically find a suitable feature collection file
-    feature_files = [
-        file
-        for file in os.listdir(data_dir)
-        if file.endswith(".json") and file != "collection.json"
-    ]
-    if not feature_files:
+    with os.scandir(data_dir) as entries:
+        # Attempt to dynamically find a suitable feature collection file
+        # Use the first found feature collection file
+        feature_file = next(
+            (
+                entry.path
+                for entry in entries
+                if entry.is_file()
+                and entry.name.endswith(".json")
+                and entry.name != "collection.json"
+            ),
+            None,
+        )
+
+    if feature_file is None:
         click.secho(
             "No feature collection files found in the specified directory.",
             fg="red",
             err=True,
         )
         raise click.Abort()
-    feature_collection_file = feature_files[
-        0
-    ]  # Use the first found feature collection file
-    feature_collection = load_data(data_dir, feature_collection_file)
 
-    load_collection(base_url, collection_id, data_dir)
+    feature_collection = load_data(feature_file)
+
+    load_collection(client, collection_id, data_dir)
     if use_bulk:
-        load_items_bulk_insert(base_url, collection_id, feature_collection, data_dir)
+        load_items_bulk_insert(client, collection_id, feature_collection)
     else:
-        load_items_one_by_one(base_url, collection_id, feature_collection, data_dir)
+        load_items_one_by_one(client, collection_id, feature_collection)
 
 
-def load_items_one_by_one(base_url, collection_id, feature_collection, data_dir):
+def load_items_one_by_one(
+    client: Client, collection_id: str, feature_collection: dict[str, Any]
+) -> None:
     """Load STAC items into the database one by one."""
     for feature in feature_collection["features"]:
-        try:
-            feature["collection"] = collection_id
-            resp = requests.post(
-                f"{base_url}/collections/{collection_id}/items", json=feature
-            )
-            if resp.status_code == 200:
-                click.echo(f"Status code: {resp.status_code}")
-                click.echo(f"Added item: {feature['id']}")
-            elif resp.status_code == 409:
-                click.echo(f"Status code: {resp.status_code}")
-                click.echo(f"Item: {feature['id']} already exists")
-        except requests.ConnectionError:
-            click.secho("Failed to connect", fg="red", err=True)
-
-
-def load_items_bulk_insert(base_url, collection_id, feature_collection, data_dir):
-    """Load STAC items into the database via bulk insert."""
-    try:
-        for i, _ in enumerate(feature_collection["features"]):
-            feature_collection["features"][i]["collection"] = collection_id
-        resp = requests.post(
-            f"{base_url}/collections/{collection_id}/items", json=feature_collection
-        )
+        feature["collection"] = collection_id
+        resp = client.post(f"/collections/{collection_id}/items", json=feature)
         if resp.status_code == 200:
             click.echo(f"Status code: {resp.status_code}")
-            click.echo("Bulk inserted items successfully.")
-        elif resp.status_code == 204:
-            click.echo(f"Status code: {resp.status_code}")
-            click.echo("Bulk update successful, no content returned.")
+            click.echo(f"Added item: {feature['id']}")
         elif resp.status_code == 409:
             click.echo(f"Status code: {resp.status_code}")
-            click.echo("Conflict detected, some items might already exist.")
-    except requests.ConnectionError:
-        click.secho("Failed to connect", fg="red", err=True)
+            click.echo(f"Item: {feature['id']} already exists")
+
+
+def load_items_bulk_insert(
+    client: Client, collection_id: str, feature_collection: dict[str, Any]
+) -> None:
+    """Load STAC items into the database via bulk insert."""
+    for feature in feature_collection["features"]:
+        feature["collection"] = collection_id
+    resp = client.post(f"/collections/{collection_id}/items", json=feature_collection)
+    if resp.status_code == 200:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo("Bulk inserted items successfully.")
+    elif resp.status_code == 204:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo("Bulk update successful, no content returned.")
+    elif resp.status_code == 409:
+        click.echo(f"Status code: {resp.status_code}")
+        click.echo("Conflict detected, some items might already exist.")
 
 
 @click.command()
@@ -117,9 +116,10 @@ def load_items_bulk_insert(base_url, collection_id, feature_collection, data_dir
     default="sample_data/",
     help="Directory containing collection.json and feature collection file",
 )
-def main(base_url, collection_id, use_bulk, data_dir):
+def main(base_url: str, collection_id: str, use_bulk: bool, data_dir: str) -> None:
     """Load STAC items into the database."""
-    load_items(base_url, collection_id, use_bulk, data_dir)
+    with Client(base_url=base_url) as client:
+        load_items(client, collection_id, use_bulk, data_dir)
 
 
 if __name__ == "__main__":

examples/auth/README.md

@@ -123,7 +123,7 @@ limited permissions to specific read-only endpoints.
       {"path": "/collections/{collection_id}", "method": ["GET"]},
       {"path": "/collections/{collection_id}/items", "method": ["GET"]},
       {"path": "/queryables", "method": ["GET"]},
-      {"path": "/queryables/collections/{collection_id}/queryables", "method": ["GET"]},
+      {"path": "/collections/{collection_id}/queryables", "method": ["GET"]},
       {"path": "/_mgmt/ping", "method": ["GET"]}
     ],
     "dependencies": [

examples/auth/compose.basic_auth.yml

@@ -21,7 +21,7 @@ services:
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
       - BACKEND=elasticsearch
-      - STAC_FASTAPI_ROUTE_DEPENDENCIES=[{"routes":[{"method":"*","path":"*"}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"admin","password":"admin"}]}}]},{"routes":[{"path":"/","method":["GET"]},{"path":"/conformance","method":["GET"]},{"path":"/collections/{collection_id}/items/{item_id}","method":["GET"]},{"path":"/search","method":["GET","POST"]},{"path":"/collections","method":["GET"]},{"path":"/collections/{collection_id}","method":["GET"]},{"path":"/collections/{collection_id}/items","method":["GET"]},{"path":"/queryables","method":["GET"]},{"path":"/queryables/collections/{collection_id}/queryables","method":["GET"]},{"path":"/_mgmt/ping","method":["GET"]}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"reader","password":"reader"}]}}]}]
+      - STAC_FASTAPI_ROUTE_DEPENDENCIES=[{"routes":[{"method":"*","path":"*"}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"admin","password":"admin"}]}}]},{"routes":[{"path":"/","method":["GET"]},{"path":"/conformance","method":["GET"]},{"path":"/collections/{collection_id}/items/{item_id}","method":["GET"]},{"path":"/search","method":["GET","POST"]},{"path":"/collections","method":["GET"]},{"path":"/collections/{collection_id}","method":["GET"]},{"path":"/collections/{collection_id}/items","method":["GET"]},{"path":"/queryables","method":["GET"]},{"path":"/collections/{collection_id}/queryables","method":["GET"]},{"path":"/_mgmt/ping","method":["GET"]}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"reader","password":"reader"}]}}]}]
     ports:
       - "8080:8080"
     volumes:
@@ -55,7 +55,7 @@ services:
       - ES_USE_SSL=false
       - ES_VERIFY_CERTS=false
       - BACKEND=opensearch
-      - STAC_FASTAPI_ROUTE_DEPENDENCIES=[{"routes":[{"method":"*","path":"*"}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"admin","password":"admin"}]}}]},{"routes":[{"path":"/","method":["GET"]},{"path":"/conformance","method":["GET"]},{"path":"/collections/{collection_id}/items/{item_id}","method":["GET"]},{"path":"/search","method":["GET","POST"]},{"path":"/collections","method":["GET"]},{"path":"/collections/{collection_id}","method":["GET"]},{"path":"/collections/{collection_id}/items","method":["GET"]},{"path":"/queryables","method":["GET"]},{"path":"/queryables/collections/{collection_id}/queryables","method":["GET"]},{"path":"/_mgmt/ping","method":["GET"]}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"reader","password":"reader"}]}}]}]
+      - STAC_FASTAPI_ROUTE_DEPENDENCIES=[{"routes":[{"method":"*","path":"*"}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"admin","password":"admin"}]}}]},{"routes":[{"path":"/","method":["GET"]},{"path":"/conformance","method":["GET"]},{"path":"/collections/{collection_id}/items/{item_id}","method":["GET"]},{"path":"/search","method":["GET","POST"]},{"path":"/collections","method":["GET"]},{"path":"/collections/{collection_id}","method":["GET"]},{"path":"/collections/{collection_id}/items","method":["GET"]},{"path":"/queryables","method":["GET"]},{"path":"/collections/{collection_id}/queryables","method":["GET"]},{"path":"/_mgmt/ping","method":["GET"]}],"dependencies":[{"method":"stac_fastapi.core.basic_auth.BasicAuth","kwargs":{"credentials":[{"username":"reader","password":"reader"}]}}]}]
     ports:
       - "8082:8082"
     volumes:

stac_fastapi/core/stac_fastapi/core/base_database_logic.py

@@ -60,6 +60,18 @@ async def delete_item(
         """Delete an item from the database."""
         pass
 
+    @abc.abstractmethod
+    async def get_items_mapping(self, collection_id: str) -> Dict[str, Dict[str, Any]]:
+        """Get the mapping for the items in the collection."""
+        pass
+
+    @abc.abstractmethod
+    async def get_items_unique_values(
+        self, collection_id: str, field_names: Iterable[str], *, limit: int = ...
+    ) -> Dict[str, List[str]]:
+        """Get the unique values for the given fields in the collection."""
+        pass
+
     @abc.abstractmethod
     async def create_collection(self, collection: Dict, refresh: bool = False) -> None:
         """Create a collection in the database."""

stac_fastapi/core/stac_fastapi/core/extensions/filter.py

@@ -60,6 +60,17 @@
         "maximum": 100,
     },
 }
+"""Queryables that are present in all collections."""
+
+OPTIONAL_QUERYABLES: Dict[str, Dict[str, Any]] = {
+    "platform": {
+        "$enum": True,
+        "description": "Satellite platform identifier",
+    },
+}
+"""Queryables that are present in some collections."""
+
+ALL_QUERYABLES: Dict[str, Dict[str, Any]] = DEFAULT_QUERYABLES | OPTIONAL_QUERYABLES
 
 
 class LogicalOp(str, Enum):

stac_fastapi/elasticsearch/setup.py

@@ -19,7 +19,6 @@
         "pytest-cov~=4.0.0",
         "pytest-asyncio~=0.21.0",
         "pre-commit~=3.0.0",
-        "requests>=2.32.0,<3.0.0",
         "ciso8601~=2.3.0",
         "httpx>=0.24.0,<0.28.0",
     ],

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py

@@ -37,6 +37,7 @@
     TokenPaginationExtension,
     TransactionExtension,
 )
+from stac_fastapi.extensions.core.filter import FilterConformanceClasses
 from stac_fastapi.extensions.third_party import BulkTransactionExtension
 from stac_fastapi.sfeos_helpers.aggregation import EsAsyncBaseAggregationClient
 from stac_fastapi.sfeos_helpers.filter import EsAsyncBaseFiltersClient
@@ -56,7 +57,7 @@
     client=EsAsyncBaseFiltersClient(database=database_logic)
 )
 filter_extension.conformance_classes.append(
-    "http://www.opengis.net/spec/cql2/1.0/conf/advanced-comparison-operators"
+    FilterConformanceClasses.ADVANCED_COMPARISON_OPERATORS
 )
 
 aggregation_extension = AggregationExtension(
@@ -103,22 +104,24 @@
 
 post_request_model = create_post_request_model(search_extensions)
 
-api = StacApi(
-    title=os.getenv("STAC_FASTAPI_TITLE", "stac-fastapi-elasticsearch"),
-    description=os.getenv("STAC_FASTAPI_DESCRIPTION", "stac-fastapi-elasticsearch"),
-    api_version=os.getenv("STAC_FASTAPI_VERSION", "5.0.0a1"),
-    settings=settings,
-    extensions=extensions,
-    client=CoreClient(
+app_config = {
+    "title": os.getenv("STAC_FASTAPI_TITLE", "stac-fastapi-elasticsearch"),
+    "description": os.getenv("STAC_FASTAPI_DESCRIPTION", "stac-fastapi-elasticsearch"),
+    "api_version": os.getenv("STAC_FASTAPI_VERSION", "5.0.0a1"),
+    "settings": settings,
+    "extensions": extensions,
+    "client": CoreClient(
         database=database_logic,
         session=session,
         post_request_model=post_request_model,
         landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
     ),
-    search_get_request_model=create_get_request_model(search_extensions),
-    search_post_request_model=post_request_model,
-    route_dependencies=get_route_dependencies(),
-)
+    "search_get_request_model": create_get_request_model(search_extensions),
+    "search_post_request_model": post_request_model,
+    "route_dependencies": get_route_dependencies(),
+}
+
+api = StacApi(**app_config)
 
 
 @asynccontextmanager

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -1038,6 +1038,37 @@ async def get_items_mapping(self, collection_id: str) -> Dict[str, Any]:
         except ESNotFoundError:
             raise NotFoundError(f"Mapping for index {index_name} not found")
 
+    async def get_items_unique_values(
+        self, collection_id: str, field_names: Iterable[str], *, limit: int = 100
+    ) -> Dict[str, List[str]]:
+        """Get the unique values for the given fields in the collection."""
+        limit_plus_one = limit + 1
+        index_name = index_alias_by_collection_id(collection_id)
+
+        query = await self.client.search(
+            index=index_name,
+            body={
+                "size": 0,
+                "aggs": {
+                    field: {"terms": {"field": field, "size": limit_plus_one}}
+                    for field in field_names
+                },
+            },
+        )
+
+        result: Dict[str, List[str]] = {}
+        for field, agg in query["aggregations"].items():
+            if len(agg["buckets"]) > limit:
+                logger.warning(
+                    "Skipping enum field %s: exceeds limit of %d unique values. "
+                    "Consider excluding this field from enumeration or increase the limit.",
+                    field,
+                    limit,
+                )
+                continue
+            result[field] = [bucket["key"] for bucket in agg["buckets"]]
+        return result
+
     async def create_collection(self, collection: Collection, **kwargs: Any):
         """Create a single collection in the database.
 

stac_fastapi/opensearch/setup.py

@@ -20,7 +20,6 @@
         "pytest-cov~=4.0.0",
         "pytest-asyncio~=0.21.0",
         "pre-commit~=3.0.0",
-        "requests>=2.32.0,<3.0.0",
         "ciso8601~=2.3.0",
         "httpx>=0.24.0,<0.28.0",
     ],