Aggregation of queryables across select collections

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Timeout option added to `Client.open` [#463](https://github.com/stac-utils/pystac-client/pull/463)
 - Support for fetching catalog queryables [#477](https://github.com/stac-utils/pystac-client/pull/477)
 - PySTAC Client specific warnings [#480](https://github.com/stac-utils/pystac-client/pull/480)
+- Support for fetching and merging a selection of queryables [#511](https://github.com/stac-utils/pystac-client/pull/511)
 
 ### Changed
 

pystac_client/client.py

@@ -1,17 +1,17 @@
-from functools import lru_cache
 import re
+import warnings
+from functools import lru_cache
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
-    cast,
     Dict,
     Iterator,
     List,
     Optional,
     Union,
+    cast,
 )
-import warnings
 
 import pystac
 import pystac.utils
@@ -22,7 +22,6 @@
 from pystac_client._utils import Modifiable, call_modifier
 from pystac_client.collection_client import CollectionClient
 from pystac_client.conformance import ConformanceClasses
-
 from pystac_client.errors import ClientTypeError
 from pystac_client.exceptions import APIError
 from pystac_client.item_search import (
@@ -39,13 +38,9 @@
     QueryLike,
     SortbyLike,
 )
-from pystac_client.mixins import QueryablesMixin
+from pystac_client.mixins import QUERYABLES_ENDPOINT, QueryablesMixin
 from pystac_client.stac_api_io import StacApiIO, Timeout
-from pystac_client.warnings import (
-    DoesNotConformTo,
-    FallbackToPystac,
-    NoConformsTo,
-)
+from pystac_client.warnings import DoesNotConformTo, FallbackToPystac, NoConformsTo
 
 if TYPE_CHECKING:
     from pystac.item import Item as Item_Type
@@ -340,17 +335,60 @@ def _warn_about_fallback(self, *args: str) -> None:
             warnings.warn(DoesNotConformTo(*args), stacklevel=2)
         warnings.warn(FallbackToPystac(), stacklevel=2)
 
+    def get_merged_queryables(self, collections: List[str]) -> Dict[str, Any]:
+        """Return the set of queryables in common to the specified collections.
+
+        Queryables from multiple collections are unioned together, except in the case
+        when the same queryable key has a different definition, in which case that key
+        is dropped.
+
+        Output is a dictionary that can be used in ``jsonshema.validate``
+
+        Args:
+            collections List[str]: The IDs of the collections to inspect.
+
+        Return:
+            Dict[str, Any]: Dictionary containing queryable fields
+        """
+        if not collections:
+            raise ValueError("cannot get_merged_queryables from empty Iterable")
+
+        if not self.conforms_to(ConformanceClasses.FILTER):
+            raise DoesNotConformTo(ConformanceClasses.FILTER.name)
+        response = self.get_queryables_from(
+            self._get_collection_queryables_href(collections[0])
+        )
+        response.pop("$id")
+        addl_props = response.get("additionalProperties", False)
+        for collection in collections[1:]:
+            resp = self.get_queryables_from(
+                self._get_collection_queryables_href(collection)
+            )
+
+            # additionalProperties is false if any collection doesn't support additional
+            # properties
+            addl_props &= resp.get("additionalProperties", False)
+
+            # drop queryables if their keys match, but the descriptions differ
+            for k in set(resp["properties"]).intersection(response["properties"]):
+                if resp["properties"][k] != response["properties"][k]:
+                    resp["properties"].pop(k)
+                    response["properties"].pop(k)
+            response["properties"].update(resp["properties"])
+        return response
+
     @lru_cache()
-    def get_collection(
-        self, collection_id: str
-    ) -> Optional[Union[Collection, CollectionClient]]:
+    def get_collection(self, collection_id: str) -> Union[Collection, CollectionClient]:
         """Get a single collection from this Catalog/API
 
         Args:
             collection_id: The Collection ID to get
 
         Returns:
             Union[Collection, CollectionClient]: A STAC Collection
+
+        Raises:
+            NotFoundError if collection_id does not exist.
         """
         collection: Union[Collection, CollectionClient]
 
@@ -602,3 +640,9 @@ def _collections_href(self, collection_id: Optional[str] = None) -> str:
         if collection_id is not None:
             return f"{href.rstrip('/')}/{collection_id}"
         return href
+
+    def _get_collection_queryables_href(
+        self, collection_id: Optional[str] = None
+    ) -> str:
+        href = self._collections_href(collection_id)
+        return f"{href.rstrip('/')}/{QUERYABLES_ENDPOINT}"

pystac_client/mixins.py

@@ -1,10 +1,10 @@
-from typing import Optional, Dict, Any, Union
 import warnings
+from typing import Any, Dict, Optional, Union
 
 import pystac
 
-from pystac_client.exceptions import APIError
 from pystac_client.conformance import ConformanceClasses
+from pystac_client.exceptions import APIError
 from pystac_client.stac_api_io import StacApiIO
 from pystac_client.warnings import DoesNotConformTo, MissingLink
 
@@ -32,22 +32,21 @@ def _get_href(self, rel: str, link: Optional[pystac.Link], endpoint: str) -> str
 class QueryablesMixin(BaseMixin):
     """Mixin for adding support for /queryables endpoint"""
 
-    def get_queryables(self) -> Dict[str, Any]:
+    def get_queryables_from(self, url: str) -> Dict[str, Any]:
         """Return all queryables.
 
         Output is a dictionary that can be used in ``jsonshema.validate``
 
+        Args:
+            url: a queryables url
+
         Return:
             Dict[str, Any]: Dictionary containing queryable fields
         """
+
         if self._stac_io is None:
             raise APIError("API access is not properly configured")
 
-        if not self.conforms_to(ConformanceClasses.FILTER):
-            raise DoesNotConformTo(ConformanceClasses.FILTER.name)
-
-        url = self._get_queryables_href()
-
         result = self._stac_io.read_json(url)
         if "properties" not in result:
             raise APIError(
@@ -57,7 +56,14 @@ def get_queryables(self) -> Dict[str, Any]:
 
         return result
 
+    def get_queryables(self) -> Dict[str, Any]:
+        url = self._get_queryables_href()
+        return self.get_queryables_from(url)
+
     def _get_queryables_href(self) -> str:
+        if not self.conforms_to(ConformanceClasses.FILTER):
+            raise DoesNotConformTo(ConformanceClasses.FILTER.name)
+
         link = self.get_single_link(QUERYABLES_REL)
         href = self._get_href(QUERYABLES_REL, link, QUERYABLES_ENDPOINT)
         return href