Merge branch 'main' into add-opensearch

Update with changes from main

Author: jamesfisher-geo

.github/workflows/cicd.yml

@@ -16,7 +16,7 @@ jobs:
     services:
 
       elasticsearch_8_svc:
-        image: docker.elastic.co/elasticsearch/elasticsearch:8.10.4
+        image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
         env:
           cluster.name: stac-cluster
           node.name: es01
@@ -58,7 +58,7 @@ jobs:
 
       # Setup Python (faster than using Python container)
       - name: Setup Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
       - name: Lint code

CHANGELOG.md

@@ -9,10 +9,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+- Advanced comparison (LIKE, IN, BETWEEN) operators to the Filter extension [#178](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/178)
+
 ### Changed
 
+- Elasticsearch drivers from 7.17.9 to 8.11.0 [#169](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/169)
+
 ### Fixed
 
+- Exclude unset fields in search response [#166](https://github.com/stac-utils/stac-fastapi-elasticsearch/issues/166)
+- Upgrade stac-fastapi to v2.4.9 [#172](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/172)
+- Set correct default filter-lang for GET /search requests [#179](https://github.com/stac-utils/stac-fastapi-elasticsearch/issues/179)
 
 ## [v1.0.0]
 

docker-compose.yml

@@ -59,7 +59,7 @@ services:
 
   elasticsearch:
     container_name: es-container
-    image: docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION:-8.10.4}
+    image: docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION:-8.11.0}
     environment:
       ES_JAVA_OPTS: -Xms512m -Xmx1g
     volumes:

stac_fastapi/elasticsearch/setup.py

@@ -10,11 +10,11 @@
     "attrs",
     "pydantic[dotenv]<2",
     "stac_pydantic==2.0.*",
-    "stac-fastapi.types==2.4.8",
-    "stac-fastapi.api==2.4.8",
-    "stac-fastapi.extensions==2.4.8",
-    "elasticsearch[async]==7.17.9",
-    "elasticsearch-dsl==7.4.1",
+    "stac-fastapi.types==2.4.9",
+    "stac-fastapi.api==2.4.9",
+    "stac-fastapi.extensions==2.4.9",
+    "elasticsearch[async]==8.11.0",
+    "elasticsearch-dsl==8.11.0",
     "pystac[validation]",
     "uvicorn",
     "orjson",

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py

@@ -24,6 +24,11 @@
 settings = ElasticsearchSettings()
 session = Session.create_from_settings(settings)
 
+filter_extension = FilterExtension(client=EsAsyncBaseFiltersClient())
+filter_extension.conformance_classes.append(
+    "http://www.opengis.net/spec/cql2/1.0/conf/advanced-comparison-operators"
+)
+
 extensions = [
     TransactionExtension(client=TransactionsClient(session=session), settings=settings),
     BulkTransactionExtension(client=BulkTransactionsClient(session=session)),
@@ -32,7 +37,7 @@
     SortExtension(),
     TokenPaginationExtension(),
     ContextExtension(),
-    FilterExtension(client=EsAsyncBaseFiltersClient()),
+    filter_extension,
 ]
 
 post_request_model = create_post_request_model(extensions)

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/config.py

@@ -1,30 +1,43 @@
 """API configuration."""
 import os
+import ssl
 from typing import Any, Dict, Set
 
 from elasticsearch import AsyncElasticsearch, Elasticsearch  # type: ignore
 from stac_fastapi.types.config import ApiSettings
 
 
 def _es_config() -> Dict[str, Any]:
+    # Determine the scheme (http or https)
+    use_ssl = os.getenv("ES_USE_SSL", "true").lower() == "true"
+    scheme = "https" if use_ssl else "http"
+
+    # Configure the hosts parameter with the correct scheme
+    hosts = [f"{scheme}://{os.getenv('ES_HOST')}:{os.getenv('ES_PORT')}"]
+
+    # Initialize the configuration dictionary
     config = {
-        "hosts": [{"host": os.getenv("ES_HOST"), "port": os.getenv("ES_PORT")}],
+        "hosts": hosts,
         "headers": {"accept": "application/vnd.elasticsearch+json; compatible-with=7"},
-        "use_ssl": True,
-        "verify_certs": True,
     }
 
-    if (u := os.getenv("ES_USER")) and (p := os.getenv("ES_PASS")):
-        config["http_auth"] = (u, p)
+    # Explicitly exclude SSL settings when not using SSL
+    if not use_ssl:
+        return config
 
-    if (v := os.getenv("ES_USE_SSL")) and v == "false":
-        config["use_ssl"] = False
+    # Include SSL settings if using https
+    config["ssl_version"] = ssl.TLSVersion.TLSv1_3  # type: ignore
+    config["verify_certs"] = os.getenv("ES_VERIFY_CERTS", "true").lower() != "false"  # type: ignore
 
-    if (v := os.getenv("ES_VERIFY_CERTS")) and v == "false":
-        config["verify_certs"] = False
+    # Include CA Certificates if verifying certs
+    if config["verify_certs"]:
+        config["ca_certs"] = os.getenv(
+            "CURL_CA_BUNDLE", "/etc/ssl/certs/ca-certificates.crt"
+        )
 
-    if v := os.getenv("CURL_CA_BUNDLE"):
-        config["ca_certs"] = v
+    # Handle authentication
+    if (u := os.getenv("ES_USER")) and (p := os.getenv("ES_PASS")):
+        config["http_auth"] = (u, p)
 
     return config
 

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -26,6 +26,7 @@
 from stac_fastapi.elasticsearch.session import Session
 from stac_fastapi.extensions.third_party.bulk_transactions import (
     BaseBulkTransactionsClient,
+    BulkTransactionMethod,
     Items,
 )
 from stac_fastapi.types import stac as stac_types
@@ -379,12 +380,12 @@ async def get_search(
             base_args["sortby"] = sort_param
 
         if filter:
-            if filter_lang == "cql2-text":
+            if filter_lang == "cql2-json":
                 base_args["filter-lang"] = "cql2-json"
-                base_args["filter"] = orjson.loads(to_cql2(parse_cql2_text(filter)))
+                base_args["filter"] = orjson.loads(unquote_plus(filter))
             else:
                 base_args["filter-lang"] = "cql2-json"
-                base_args["filter"] = orjson.loads(unquote_plus(filter))
+                base_args["filter"] = orjson.loads(to_cql2(parse_cql2_text(filter)))
 
         if fields:
             includes = set()
@@ -509,7 +510,9 @@ async def post_search(
             filter_kwargs = search_request.fields.filter_fields
 
             items = [
-                orjson.loads(stac_pydantic.Item(**feat).json(**filter_kwargs))
+                orjson.loads(
+                    stac_pydantic.Item(**feat).json(**filter_kwargs, exclude_unset=True)
+                )
                 for feat in items
             ]
 
@@ -566,7 +569,7 @@ async def create_item(
         if item["type"] == "FeatureCollection":
             bulk_client = BulkTransactionsClient()
             processed_items = [
-                bulk_client.preprocess_item(item, base_url) for item in item["features"]  # type: ignore
+                bulk_client.preprocess_item(item, base_url, BulkTransactionMethod.INSERT) for item in item["features"]  # type: ignore
             ]
 
             await self.database.bulk_async(
@@ -716,17 +719,23 @@ def __attrs_post_init__(self):
         settings = ElasticsearchSettings()
         self.client = settings.create_client
 
-    def preprocess_item(self, item: stac_types.Item, base_url) -> stac_types.Item:
+    def preprocess_item(
+        self, item: stac_types.Item, base_url, method: BulkTransactionMethod
+    ) -> stac_types.Item:
         """Preprocess an item to match the data model.
 
         Args:
             item: The item to preprocess.
             base_url: The base URL of the request.
+            method: The bulk transaction method.
 
         Returns:
             The preprocessed item.
         """
-        return self.database.sync_prep_create_item(item=item, base_url=base_url)
+        exist_ok = method == BulkTransactionMethod.UPSERT
+        return self.database.sync_prep_create_item(
+            item=item, base_url=base_url, exist_ok=exist_ok
+        )
 
     @overrides
     def bulk_item_insert(
@@ -749,7 +758,8 @@ def bulk_item_insert(
             base_url = ""
 
         processed_items = [
-            self.preprocess_item(item, base_url) for item in items.items.values()
+            self.preprocess_item(item, base_url, items.method)
+            for item in items.items.values()
         ]
 
         # not a great way to get the collection_id-- should be part of the method signature

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_elasticsearch/database_logic.py

@@ -171,18 +171,19 @@ def indices(collection_ids: Optional[List[str]]) -> str:
 
 
 async def create_collection_index() -> None:
-    """Create the index for Collections in Elasticsearch.
+    """
+    Create the index for a Collection.
+
+    Returns:
+        None
 
-    This function creates the Elasticsearch index for the `Collections` with the predefined mapping.
-    If the index already exists, the function ignores the error and continues execution.
     """
     client = AsyncElasticsearchSettings().create_client
 
-    await client.indices.create(
+    await client.options(ignore_status=400).indices.create(
         index=f"{COLLECTIONS_INDEX}-000001",
         aliases={COLLECTIONS_INDEX: {}},
         mappings=ES_COLLECTIONS_MAPPINGS,
-        ignore=400,  # ignore 400 already exists code
     )
     await client.close()
 
@@ -201,12 +202,11 @@ async def create_item_index(collection_id: str):
     client = AsyncElasticsearchSettings().create_client
     index_name = index_by_collection_id(collection_id)
 
-    await client.indices.create(
+    await client.options(ignore_status=400).indices.create(
         index=f"{index_by_collection_id(collection_id)}-000001",
         aliases={index_name: {}},
         mappings=ES_ITEMS_MAPPINGS,
         settings=ES_ITEMS_SETTINGS,
-        ignore=400,  # ignore 400 already exists code
     )
     await client.close()
 
@@ -577,13 +577,16 @@ async def check_collection_exists(self, collection_id: str):
         if not await self.client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise NotFoundError(f"Collection {collection_id} does not exist")
 
-    async def prep_create_item(self, item: Item, base_url: str) -> Item:
+    async def prep_create_item(
+        self, item: Item, base_url: str, exist_ok: bool = False
+    ) -> Item:
         """
         Preps an item for insertion into the database.
 
         Args:
             item (Item): The item to be prepped for insertion.
             base_url (str): The base URL used to create the item's self URL.
+            exist_ok (bool): Indicates whether the item can exist already.
 
         Returns:
             Item: The prepped item.
@@ -594,7 +597,7 @@ async def prep_create_item(self, item: Item, base_url: str) -> Item:
         """
         await self.check_collection_exists(collection_id=item["collection"])
 
-        if await self.client.exists(
+        if not exist_ok and await self.client.exists(
             index=index_by_collection_id(item["collection"]),
             id=mk_item_id(item["id"], item["collection"]),
         ):
@@ -604,17 +607,20 @@ async def prep_create_item(self, item: Item, base_url: str) -> Item:
 
         return self.item_serializer.stac_to_db(item, base_url)
 
-    def sync_prep_create_item(self, item: Item, base_url: str) -> Item:
+    def sync_prep_create_item(
+        self, item: Item, base_url: str, exist_ok: bool = False
+    ) -> Item:
         """
         Prepare an item for insertion into the database.
 
         This method performs pre-insertion preparation on the given `item`,
         such as checking if the collection the item belongs to exists,
-        and verifying that an item with the same ID does not already exist in the database.
+        and optionally verifying that an item with the same ID does not already exist in the database.
 
         Args:
             item (Item): The item to be inserted into the database.
             base_url (str): The base URL used for constructing URLs for the item.
+            exist_ok (bool): Indicates whether the item can exist already.
 
         Returns:
             Item: The item after preparation is done.
@@ -628,7 +634,7 @@ def sync_prep_create_item(self, item: Item, base_url: str) -> Item:
         if not self.sync_client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise NotFoundError(f"Collection {collection_id} does not exist")
 
-        if self.sync_client.exists(
+        if not exist_ok and self.sync_client.exists(
             index=index_by_collection_id(collection_id),
             id=mk_item_id(item_id, collection_id),
         ):