Coerce cql2 style to match HTTP method (#804)

Use cql2 library to convert between cql2-text and cql2-json

Author: jsignell

CHANGELOG.md

@@ -7,9 +7,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+
+- Coerce cql2 style to match HTTP method using `cql2` library ([#804](https://github.com/stac-utils/pystac-client/pull/804))
+
 ### Fixed
 
-- Fix usage documentation of `ItemSearch`
+- Fix usage documentation of `ItemSearch` ([#790](https://github.com/stac-utils/pystac-client/pull/790))
 - Fix fields argument to CLI  ([#797](https://github.com/stac-utils/pystac-client/pull/797))
 - Clarify recursive behaviour of the `get_items` method in the method docstring ([#800](https://github.com/stac-utils/pystac-client/pull/800))
 

pyproject.toml

@@ -39,6 +39,7 @@ stac-client = "pystac_client.cli:cli"
 dev = [
     "codespell~=2.4.0",
     "coverage~=7.2",
+    "cql2>=0.3.7",
     "doc8~=1.1.1",
     "importlib-metadata~=8.0",
     "mypy~=1.2",
@@ -55,7 +56,7 @@ dev = [
     "tomli~=2.0; python_version<'3.11'",
     "types-python-dateutil>=2.8.19,<2.10.0",
     "types-requests~=2.32.0",
-    "urllib3>=2.0,<2.3.0",                   # v2.3.0 breaks VCR, b/c https://github.com/urllib3/urllib3/pull/3489
+    "urllib3>=2.0,<2.3.0", # v2.3.0 breaks VCR, b/c https://github.com/urllib3/urllib3/pull/3489
 ]
 docs = [
     "Sphinx~=8.0",

pystac_client/item_search.py

@@ -159,6 +159,7 @@ def __init__(
 
         self.method = method
         self.modifier = modifier
+
         params = {
             "limit": limit,
             "bbox": self._format_bbox(bbox),
@@ -167,8 +168,8 @@ def __init__(
             "collections": self._format_collections(collections),
             "intersects": self._format_intersects(intersects),
             "query": self._format_query(query),
-            "filter": self._format_filter(filter),
-            "filter-lang": self._format_filter_lang(filter, filter_lang),
+            "filter": self._format_filter(method, filter_lang, filter),
+            "filter-lang": self._format_filter_lang(method, filter, filter_lang),
             "sortby": self._format_sortby(sortby),
             "fields": self._format_fields(fields),
             "q": q,
@@ -204,6 +205,8 @@ def _clean_params_for_get_request(self) -> dict[str, Any]:
             params["sortby"] = self._sortby_dict_to_str(params["sortby"])
         if "fields" in params:
             params["fields"] = self._fields_dict_to_str(params["fields"])
+        if "filter" in params and isinstance(params["filter"], dict):
+            params["filter"] = json.dumps(params["filter"])
         return params
 
     def url_with_parameters(self) -> str:
@@ -266,29 +269,69 @@ def _format_query(self, value: QueryLike | None) -> dict[str, Any] | None:
 
     @staticmethod
     def _format_filter_lang(
-        _filter: FilterLike | None, value: FilterLangLike | None
+        method: str | None,
+        _filter: FilterLike | None,
+        value: FilterLangLike | None,
     ) -> str | None:
         if _filter is None:
             return None
 
         if value is not None:
             return value
 
-        if isinstance(_filter, str):
+        if method == "GET":
             return "cql2-text"
 
-        if isinstance(_filter, dict):
+        if method == "POST":
             return "cql2-json"
 
         return None
 
-    def _format_filter(self, value: FilterLike | None) -> FilterLike | None:
-        if value is None:
+    def _format_filter(
+        self,
+        method: str | None,
+        filter_lang: FilterLangLike | None,
+        value: FilterLike | None,
+    ) -> FilterLike | None:
+        if not value:
             return None
 
         if self.client and not self.client.conforms_to(ConformanceClasses.FILTER):
             warnings.warn(DoesNotConformTo("FILTER"))
 
+        if method == "GET" and isinstance(value, str):
+            return value
+
+        if method == "POST" and isinstance(value, dict):
+            return value
+
+        # if filter_lang is specified, do not coerce
+        if filter_lang is not None:
+            return value
+
+        try:
+            import cql2
+
+            if isinstance(value, dict):
+                expr = cql2.parse_json(json.dumps(value))
+            else:
+                # could be cql2-text or stringified cql2-json
+                expr = cql2.Expr(value)
+
+        except ImportError as e:
+            raise ValueError(
+                "Unless you specify ``filter_lang`` pystac-client will try to convert "
+                "the filter to cql2-text or cql2-json based on the HTTP method "
+                "provided.\n"
+                "Resolve this error by installing ``cql2``: ``pip install cql2``"
+            ) from e
+
+        if method == "GET":
+            return str(expr.to_text())
+
+        if method == "POST":
+            return dict(expr.to_json())
+
         return value
 
     @staticmethod

tests/test_base_search.py

@@ -330,14 +330,14 @@ def test_intersects_non_geo_interface_object(self) -> None:
         with pytest.raises(Exception):
             BaseSearch(url=SEARCH_URL, intersects=object())  # type: ignore
 
-    def test_filter_lang_default_for_dict(self) -> None:
-        search = BaseSearch(url=SEARCH_URL, filter={})
-        assert search.get_parameters()["filter-lang"] == "cql2-json"
-
-    def test_filter_lang_default_for_str(self) -> None:
-        search = BaseSearch(url=SEARCH_URL, filter="")
+    def test_filter_lang_default_for_method_despite_filter_as_dict(self) -> None:
+        search = BaseSearch(url=SEARCH_URL, method="GET", filter={})
         assert search.get_parameters()["filter-lang"] == "cql2-text"
 
+    def test_filter_lang_default_for_method_despite_filter_as_str(self) -> None:
+        search = BaseSearch(url=SEARCH_URL, method="POST", filter="")
+        assert search.get_parameters()["filter-lang"] == "cql2-json"
+
     def test_filter_lang_cql2_text(self) -> None:
         # Use specified filter_lang
         search = BaseSearch(url=SEARCH_URL, filter_lang="cql2-text", filter={})
@@ -353,6 +353,51 @@ def test_filter_lang_without_filter(self) -> None:
         search = BaseSearch(url=SEARCH_URL)
         assert "filter-lang" not in search.get_parameters()
 
+    def test_filter_conversion_to_cql2_json(self) -> None:
+        search = BaseSearch(url=SEARCH_URL, method="POST", filter="eo:cloud_cover<=10")
+        assert search.get_parameters()["filter-lang"] == "cql2-json"
+        assert search.get_parameters()["filter"] == {
+            "args": [{"property": "eo:cloud_cover"}, 10],
+            "op": "<=",
+        }
+
+    def test_filter_conversion_to_cql2_text(self) -> None:
+        search = BaseSearch(
+            url=SEARCH_URL,
+            method="GET",
+            filter={"op": "<=", "args": [{"property": "eo:cloud_cover"}, 10]},
+        )
+        assert search.get_parameters()["filter-lang"] == "cql2-text"
+        assert search.get_parameters()["filter"] == '("eo:cloud_cover" <= 10)'
+
+    def test_filter_conversion_does_not_happen_if_filter_lang_specified_json(
+        self,
+    ) -> None:
+        search = BaseSearch(
+            url=SEARCH_URL,
+            method="GET",
+            filter={"op": "<=", "args": [{"property": "eo:cloud_cover"}, 10]},
+            filter_lang="cql2-json",
+        )
+        # assert search.get_parameters()["filter-lang"] == "cql2-json"
+        assert (
+            search.get_parameters()["filter"]
+            == '{"op": "<=", "args": [{"property": "eo:cloud_cover"}, 10]}'
+        )
+
+    def test_filter_conversion_does_not_happen_if_filter_lang_specified_text(
+        self,
+    ) -> None:
+        search = BaseSearch(
+            url=SEARCH_URL,
+            method="POST",
+            filter="eo:cloud_cover<=10",
+            filter_lang="cql2-text",
+        )
+        # note that this is likely to fail when it hits the server
+        assert search.get_parameters()["filter-lang"] == "cql2-text"
+        assert search.get_parameters()["filter"] == "eo:cloud_cover<=10"
+
     def test_sortby(self) -> None:
         search = BaseSearch(url=SEARCH_URL, sortby="properties.datetime")
         assert search.get_parameters()["sortby"] == [