move Client-specific get_queryables implementation to Client class

pystac_client/client.py

@@ -335,6 +335,54 @@ def _warn_about_fallback(self, *args: str) -> None:
             warnings.warn(DoesNotConformTo(*args), stacklevel=2)
         warnings.warn(FallbackToPystac(), stacklevel=2)
 
+    def get_queryables(self, *collections: Optional[str]) -> Dict[str, Any]:
+        """Return all queryables, or limit to those of 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: The IDs of the items to include.
+
+        Return:
+            Dict[str, Any]: Dictionary containing queryable fields
+        """
+        if not collections:
+            return super().get_queryables()
+
+        col = self.get_collection(collections[0])
+        assert isinstance(col, CollectionClient)
+        response = col.get_queryables()
+        addProps = response.get("additionalProperties", None)
+        for collection in collections[1:]:
+            col = self.get_collection(collection)
+            assert isinstance(col, CollectionClient)
+            col_resp = col.get_queryables()
+
+            # Ensure schema and additionalProperties consistency
+            assert response["$schema"] == col_resp["$schema"], (
+                f"$schema inconsistency between {response['$id']} "
+                f"and {col_resp['$id']} ('{response['$schema']}' "
+                f"!= '{col_resp['$schema']}')."
+            )
+            assert addProps == col_resp.get("additionalProperties", None), (
+                "'additionalProperties' varies across collections specified"
+                f" ({collections})"
+            )
+
+            # drop queryables if their keys match, but the descriptions differ
+            for k in set(col_resp["properties"].keys()).intersection(
+                response["properties"].keys()
+            ):
+                if col_resp["properties"][k] != response["properties"][k]:
+                    col_resp["properties"].pop(k)
+                    response["properties"].pop(k)
+            response["properties"].update(col_resp["properties"])
+
+        return response
+
     @lru_cache()
     def get_collection(self, collection_id: str) -> Union[Collection, CollectionClient]:
         """Get a single collection from this Catalog/API

pystac_client/mixins.py

@@ -32,26 +32,14 @@ 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, *collections: Optional[str]) -> Dict[str, Any]:
-        """Return all queryables, or limit to those of 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.
+    def get_queryables(self) -> Dict[str, Any]:
+        """Return all queryables.
 
         Output is a dictionary that can be used in ``jsonshema.validate``
-        Args:
-            *collections: The IDs of the items to include.
 
         Return:
             Dict[str, Any]: Dictionary containing queryable fields
         """
-        if collections and isinstance(self, pystac.Catalog):
-            response = self.get_collection(collections[0]).get_queryables()
-            response.pop("$id")
-            for collection in collections[1:]:
-                col_resp = self.get_collection(collection).get_queryables()
-                response["properties"].update(col_resp["properties"])
-            return response
-
         if self._stac_io is None:
             raise APIError("API access is not properly configured")