Implement Filter Extension (#100)

* Filter Extension implemenation

Author: Phil Varner

CHANGELOG.md

@@ -11,16 +11,23 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+- Filter Extension as GET with CQL2-Text and POST with CQL2-JSON, 
+  supporting the Basic CQL2 and Basic Spatial Operators conformance classes.
+- Added Elasticsearch local config to support snapshot/restore to local filesystem
+
 ### Fixed
 
-- Fixed search intersects query
+- Fixed search intersects query.
+- Corrected the Sort and Query conformance class URIs.
 
 ### Changed
 
 - Default to Python 3.10
 - Default to Elasticsearch 8.x
 - Collection objects are now stored in `collections` index rather than `stac_collections` index
 - Item objects are no longer stored in `stac_items`, but in indices per collection named `items_{collection_id}`
+- When using bulk ingest, items will continue to be ingested if any of them fail. Previously, the call would fail
+  immediately if any items failed.
 
 ### Removed
 

stac_fastapi/elasticsearch/setup.py

@@ -19,6 +19,8 @@
     "uvicorn",
     "overrides",
     "starlette",
+    "geojson-pydantic",
+    "pygeofilter==0.1.2",
 ]
 
 extra_reqs = {

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py

@@ -9,13 +9,15 @@
 from stac_fastapi.elasticsearch.core import (
     BulkTransactionsClient,
     CoreClient,
+    EsAsyncBaseFiltersClient,
     TransactionsClient,
 )
 from stac_fastapi.elasticsearch.database_logic import create_collection_index
 from stac_fastapi.elasticsearch.extensions import QueryExtension
 from stac_fastapi.elasticsearch.session import Session
 from stac_fastapi.extensions.core import (  # FieldsExtension,
     ContextExtension,
+    FilterExtension,
     SortExtension,
     TokenPaginationExtension,
     TransactionExtension,
@@ -26,17 +28,33 @@
 session = Session.create_from_settings(settings)
 
 
-# All of these extensions have their conformance class URL
-# incorrect, with an extra `/` before the #
 @attr.s
 class FixedSortExtension(SortExtension):
-    """Fixed Sort Extension string."""
+    """SortExtension class fixed with correct paths, removing extra forward-slash."""
 
     conformance_classes: List[str] = attr.ib(
         factory=lambda: ["https://api.stacspec.org/v1.0.0-beta.4/item-search#sort"]
     )
 
 
+@attr.s
+class FixedFilterExtension(FilterExtension):
+    """FilterExtension class fixed with correct paths, removing extra forward-slash."""
+
+    conformance_classes: List[str] = attr.ib(
+        default=[
+            "https://api.stacspec.org/v1.0.0-rc.1/item-search#filter",
+            "http://www.opengis.net/spec/ogcapi-features-3/1.0/conf/filter",
+            "http://www.opengis.net/spec/ogcapi-features-3/1.0/conf/features-filter",
+            "http://www.opengis.net/spec/cql2/1.0/conf/cql2-text",
+            "http://www.opengis.net/spec/cql2/1.0/conf/cql2-json",
+            "http://www.opengis.net/spec/cql2/1.0/conf/basic-cql2",
+            "http://www.opengis.net/spec/cql2/1.0/conf/basic-spatial-operators",
+        ]
+    )
+    client = attr.ib(factory=EsAsyncBaseFiltersClient)
+
+
 @attr.s
 class FixedQueryExtension(QueryExtension):
     """Fixed Query Extension string."""
@@ -54,6 +72,7 @@ class FixedQueryExtension(QueryExtension):
     FixedSortExtension(),
     TokenPaginationExtension(),
     ContextExtension(),
+    FixedFilterExtension(),
 ]
 
 post_request_model = create_post_request_model(extensions)

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -3,7 +3,7 @@
 import logging
 from datetime import datetime as datetime_type
 from datetime import timezone
-from typing import List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type, Union
 from urllib.parse import urljoin
 
 import attr
@@ -21,12 +21,17 @@
 from stac_fastapi.elasticsearch.models.links import PagingLinks
 from stac_fastapi.elasticsearch.serializers import CollectionSerializer, ItemSerializer
 from stac_fastapi.elasticsearch.session import Session
+from stac_fastapi.extensions.core.filter.request import FilterLang
 from stac_fastapi.extensions.third_party.bulk_transactions import (
     BaseBulkTransactionsClient,
     Items,
 )
 from stac_fastapi.types import stac as stac_types
-from stac_fastapi.types.core import AsyncBaseCoreClient, AsyncBaseTransactionsClient
+from stac_fastapi.types.core import (
+    AsyncBaseCoreClient,
+    AsyncBaseFiltersClient,
+    AsyncBaseTransactionsClient,
+)
 from stac_fastapi.types.links import CollectionLinks
 from stac_fastapi.types.stac import Collection, Collections, Item, ItemCollection
 
@@ -172,6 +177,8 @@ async def get_search(
         token: Optional[str] = None,
         fields: Optional[List[str]] = None,
         sortby: Optional[str] = None,
+        # filter: Optional[str] = None, # todo: requires fastapi > 2.3 unreleased
+        # filter_lang: Optional[str] = None, # todo: requires fastapi > 2.3 unreleased
         **kwargs,
     ) -> ItemCollection:
         """GET search catalog."""
@@ -183,8 +190,10 @@ async def get_search(
             "token": token,
             "query": json.loads(query) if query else query,
         }
+
         if datetime:
             base_args["datetime"] = datetime
+
         if sortby:
             # https://github.com/radiantearth/stac-spec/tree/master/api-spec/extensions/sort#http-get-or-post-form
             sort_param = []
@@ -197,6 +206,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 fields:
         #     includes = set()
         #     excludes = set()
@@ -264,6 +280,15 @@ async def post_search(
                         search=search, op=op, field=field, value=value
                     )
 
+        filter_lang = getattr(search_request, "filter_lang", None)
+
+        if hasattr(search_request, "filter"):
+            cql2_filter = getattr(search_request, "filter", None)
+            if filter_lang in [None, FilterLang.cql2_json]:
+                search = self.database.apply_cql2_filter(search, cql2_filter)
+            else:
+                raise Exception("CQL2-Text is not supported with POST")
+
         sort = None
         if search_request.sortby:
             sort = self.database.populate_sort(search_request.sortby)
@@ -455,3 +480,68 @@ def bulk_item_insert(
         )
 
         return f"Successfully added {len(processed_items)} Items."
+
+
+@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
+    async def get_queryables(
+        self, collection_id: Optional[str] = None, **kwargs
+    ) -> Dict[str, Any]:
+        """Get the queryables available for the given collection_id.
+
+        If collection_id is None, returns the intersection of all
+        queryables over all collections.
+
+        This base implementation returns a blank queryable schema. This is not allowed
+        under OGC CQL but it is allowed by the STAC API Filter Extension
+
+        https://github.com/radiantearth/stac-api-spec/tree/master/fragments/filter#queryables
+        """
+        return {
+            "$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#/id",
+                },
+                "collection": {
+                    "description": "Collection",
+                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/collection",
+                },
+                "geometry": {
+                    "description": "Geometry",
+                    "$ref": "https://schemas.stacspec.org/v1.0.0/item-spec/json-schema/item.json#/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,
+                },
+            },
+            "additionalProperties": True,
+        }

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -13,6 +13,7 @@
 from stac_fastapi.elasticsearch.config import (
     ElasticsearchSettings as SyncElasticsearchSettings,
 )
+from stac_fastapi.elasticsearch.extensions import filter
 from stac_fastapi.types.errors import ConflictError, NotFoundError
 from stac_fastapi.types.stac import Collection, Item
 
@@ -312,6 +313,13 @@ def apply_stacql_filter(search: Search, op: str, field: str, value: float):
 
         return search
 
+    @staticmethod
+    def apply_cql2_filter(search: Search, _filter: Optional[Dict[str, Any]]):
+        """Database logic to perform query for search endpoint."""
+        if _filter is not None:
+            search = search.filter(filter.Clause.parse_obj(_filter).to_es())
+        return search
+
     @staticmethod
     def populate_sort(sortby: List) -> Optional[Dict[str, Dict[str, str]]]:
         """Database logic to sort search instance."""