Merge branch 'main' into collection-pagination

Author: jonhealy1

CHANGELOG.md

@@ -11,14 +11,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - Collection-level Assets to the CollectionSerializer [#148](https://github.com/stac-utils/stac-fastapi-elasticsearch/issues/148)
 - Pagination for /collections - GET all collections - route [#164](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/164)
-
-### Added
-
 - Examples folder with example docker setup for running sfes from pip [#147](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/147)
+- GET /search filter extension queries [#163](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/163)
+- Added support for GET /search intersection queries [#158](https://github.com/stac-utils/stac-fastapi-elasticsearch/issues/158)
 
 ### Changed
 
 - Update elasticsearch version from 8.1.3 to 8.10.4 in cicd, gh actions [#164](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/164)
+- Updated core stac-fastapi libraries to 2.4.8 from 2.4.3 [#151](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/151)
+- Use aliases on Elasticsearch indices, add number suffix in index name. [#152](https://github.com/stac-utils/stac-fastapi-elasticsearch/pull/152)
 
 ### Fixed
 

stac_fastapi/elasticsearch/setup.py

@@ -17,6 +17,7 @@
     "elasticsearch-dsl==7.4.1",
     "pystac[validation]",
     "uvicorn",
+    "orjson",
     "overrides",
     "starlette",
     "geojson-pydantic",

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -1,20 +1,22 @@
 """Item crud client."""
-import json
 import logging
 from base64 import urlsafe_b64encode
+import re
 from datetime import datetime as datetime_type
 from datetime import timezone
 from typing import Any, Dict, List, Optional, Set, Type, Union
-from urllib.parse import urljoin
+from urllib.parse import unquote_plus, urljoin
 
 import attr
+import orjson
 import stac_pydantic
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from overrides import overrides
 from pydantic import ValidationError
+from pygeofilter.backends.cql2_json import to_cql2
+from pygeofilter.parsers.cql2_text import parse as parse_cql2_text
 from stac_pydantic.links import Relations
 from stac_pydantic.shared import MimeTypes
-from starlette.requests import Request
 
 from stac_fastapi.elasticsearch import serializers
 from stac_fastapi.elasticsearch.config import ElasticsearchSettings
@@ -303,9 +305,9 @@ def _return_date(interval_str):
 
         return {"lte": end_date, "gte": start_date}
 
-    @overrides
     async def get_search(
         self,
+        request: Request,
         collections: Optional[List[str]] = None,
         ids: Optional[List[str]] = None,
         bbox: Optional[List[NumType]] = None,
@@ -316,8 +318,8 @@ async def get_search(
         fields: Optional[List[str]] = None,
         sortby: Optional[str] = None,
         intersects: Optional[str] = None,
-        # filter: Optional[str] = None, # todo: requires fastapi > 2.3 unreleased
-        # filter_lang: Optional[str] = None, # todo: requires fastapi > 2.3 unreleased
+        filter: Optional[str] = None,
+        filter_lang: Optional[str] = None,
         **kwargs,
     ) -> ItemCollection:
         """Get search results from the database.
@@ -347,17 +349,24 @@ async def get_search(
             "bbox": bbox,
             "limit": limit,
             "token": token,
-            "query": json.loads(query) if query else query,
+            "query": orjson.loads(query) if query else query,
         }
 
+        # this is borrowed from stac-fastapi-pgstac
+        # Kludgy fix because using factory does not allow alias for filter-lan
+        query_params = str(request.query_params)
+        if filter_lang is None:
+            match = re.search(r"filter-lang=([a-z0-9-]+)", query_params, re.IGNORECASE)
+            if match:
+                filter_lang = match.group(1)
+
         if datetime:
             base_args["datetime"] = datetime
 
         if intersects:
-            base_args["intersects"] = intersects
+            base_args["intersects"] = orjson.loads(unquote_plus(intersects))
 
         if sortby:
-            # https://github.com/radiantearth/stac-spec/tree/master/api-spec/extensions/sort#http-get-or-post-form
             sort_param = []
             for sort in sortby:
                 sort_param.append(
@@ -368,12 +377,13 @@ async def get_search(
                 )
             base_args["sortby"] = sort_param
 
-        # todo: requires fastapi > 2.3 unreleased
-        # if filter:
-        #     if filter_lang == "cql2-text":
-        #         base_args["filter-lang"] = "cql2-json"
-        #         base_args["filter"] = orjson.loads(to_cql2(parse_cql2_text(filter)))
-        #         print(f'>>> {base_args["filter"]}')
+        if filter:
+            if filter_lang == "cql2-text":
+                base_args["filter-lang"] = "cql2-json"
+                base_args["filter"] = orjson.loads(to_cql2(parse_cql2_text(filter)))
+            else:
+                base_args["filter-lang"] = "cql2-json"
+                base_args["filter"] = orjson.loads(unquote_plus(filter))
 
         if fields:
             includes = set()
@@ -392,13 +402,12 @@ async def get_search(
             search_request = self.post_request_model(**base_args)
         except ValidationError:
             raise HTTPException(status_code=400, detail="Invalid parameters provided")
-        resp = await self.post_search(search_request, request=kwargs["request"])
+        resp = await self.post_search(search_request=search_request, request=request)
 
         return resp
 
-    @overrides
     async def post_search(
-        self, search_request: BaseSearchPostRequest, **kwargs
+        self, search_request: BaseSearchPostRequest, request: Request
     ) -> ItemCollection:
         """
         Perform a POST search on the catalog.
@@ -413,7 +422,6 @@ async def post_search(
         Raises:
             HTTPException: If there is an error with the cql2_json filter.
         """
-        request: Request = kwargs["request"]
         base_url = str(request.base_url)
 
         search = self.database.make_search()
@@ -500,7 +508,7 @@ async def post_search(
             filter_kwargs = search_request.fields.filter_fields
 
             items = [
-                json.loads(stac_pydantic.Item(**feat).json(**filter_kwargs))
+                orjson.loads(stac_pydantic.Item(**feat).json(**filter_kwargs))
                 for feat in items
             ]
 

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -39,7 +39,7 @@
     ":",
 }
 
-DEFAULT_INDICES = f"*,-*kibana*,-{COLLECTIONS_INDEX}"
+ITEM_INDICES = f"{ITEMS_INDEX_PREFIX}*,-*kibana*,-{COLLECTIONS_INDEX}*"
 
 DEFAULT_SORT = {
     "properties.datetime": {"order": "desc"},
@@ -164,7 +164,7 @@ def indices(collection_ids: Optional[List[str]]) -> str:
         A string of comma-separated index names. If `collection_ids` is None, returns the default indices.
     """
     if collection_ids is None:
-        return DEFAULT_INDICES
+        return ITEM_INDICES
     else:
         return ",".join([index_by_collection_id(c) for c in collection_ids])
 
@@ -178,7 +178,8 @@ async def create_collection_index() -> None:
     client = AsyncElasticsearchSettings().create_client
 
     await client.indices.create(
-        index=COLLECTIONS_INDEX,
+        index=f"{COLLECTIONS_INDEX}-000001",
+        aliases={COLLECTIONS_INDEX: {}},
         mappings=ES_COLLECTIONS_MAPPINGS,
         ignore=400,  # ignore 400 already exists code
     )
@@ -197,9 +198,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(
-        index=index_by_collection_id(collection_id),
+        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
@@ -215,7 +218,14 @@ async def delete_item_index(collection_id: str):
     """
     client = AsyncElasticsearchSettings().create_client
 
-    await client.indices.delete(index=index_by_collection_id(collection_id))
+    name = index_by_collection_id(collection_id)
+    resolved = await client.indices.resolve_index(name=name)
+    if "aliases" in resolved and resolved["aliases"]:
+        [alias] = resolved["aliases"]
+        await client.indices.delete_alias(index=alias["indices"], name=alias["name"])
+        await client.indices.delete(index=alias["indices"])
+    else:
+        await client.indices.delete(index=name)
     await client.close()
 
 
@@ -786,14 +796,11 @@ async def bulk_async(
             `mk_actions` function is called to generate a list of actions for the bulk insert. If `refresh` is set to True, the
             index is refreshed after the bulk insert. The function does not return any value.
         """
-        await asyncio.get_event_loop().run_in_executor(
-            None,
-            lambda: helpers.bulk(
-                self.sync_client,
-                mk_actions(collection_id, processed_items),
-                refresh=refresh,
-                raise_on_error=False,
-            ),
+        await helpers.async_bulk(
+            self.client,
+            mk_actions(collection_id, processed_items),
+            refresh=refresh,
+            raise_on_error=False,
         )
 
     def bulk_sync(
@@ -824,7 +831,7 @@ def bulk_sync(
     async def delete_items(self) -> None:
         """Danger. this is only for tests."""
         await self.client.delete_by_query(
-            index=DEFAULT_INDICES,
+            index=ITEM_INDICES,
             body={"query": {"match_all": {}}},
             wait_for_completion=True,
         )

stac_fastapi/elasticsearch/tests/api/test_api.py

@@ -255,7 +255,28 @@ async def test_search_invalid_date(app_client, ctx):
 
 
 @pytest.mark.asyncio
-async def test_search_point_intersects(app_client, ctx):
+async def test_search_point_intersects_get(app_client, ctx):
+    resp = await app_client.get(
+        '/search?intersects={"type":"Point","coordinates":[150.04,-33.14]}'
+    )
+
+    assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_search_polygon_intersects_get(app_client, ctx):
+    resp = await app_client.get(
+        '/search?intersects={"type":"Polygon","coordinates":[[[149.04, -34.14],[149.04, -32.14],[151.04, -32.14],[151.04, -34.14],[149.04, -34.14]]]}'
+    )
+
+    assert resp.status_code == 200
+    resp_json = resp.json()
+    assert len(resp_json["features"]) == 1
+
+
+async def test_search_point_intersects_post(app_client, ctx):
     point = [150.04, -33.14]
     intersects = {"type": "Point", "coordinates": point}