[WIP] Configurable options

pystac_client/_utils.py

@@ -1,9 +1,10 @@
 import warnings
-from typing import Callable, Optional, Union
+from typing import Callable, Optional, Union, Literal, cast
 
 import pystac
 
 from pystac_client.errors import IgnoredResultWarning
+from pystac_client.options import get_options, T_Keys
 
 Modifiable = Union[pystac.Collection, pystac.Item, pystac.ItemCollection, dict]
 
@@ -23,3 +24,17 @@ def call_modifier(
             "a function that returns 'None' or silence this warning.",
             IgnoredResultWarning,
         )
+
+
+def respond(
+    event: Literal["does_not_conform_to", "missing_link", "fallback_to_pystac"],
+    msg: str,
+) -> None:
+    """Response to event based on user-configured options"""
+    on_event = get_options()[cast(T_Keys, f"on_{event}")]
+    if on_event == "ignore":
+        pass
+    elif on_event == "warn":
+        warnings.warn(msg, UserWarning)
+    elif on_event == "error":
+        raise NotImplementedError(msg)

pystac_client/client.py

@@ -7,7 +7,7 @@
 from pystac import CatalogType, Collection
 from requests import Request
 
-from pystac_client._utils import Modifiable, call_modifier
+from pystac_client._utils import Modifiable, call_modifier, respond
 from pystac_client.collection_client import CollectionClient
 from pystac_client.conformance import ConformanceClasses
 from pystac_client.errors import ClientTypeError
@@ -203,14 +203,14 @@ def from_file(  # type: ignore
 
         return client
 
+    def conforms_to(self, *conformance_classes: ConformanceClasses) -> bool:
+        return bool(self._stac_io and self._stac_io.conforms_to(*conformance_classes))
+
     def _supports_collections(self) -> bool:
-        return self._conforms_to(ConformanceClasses.COLLECTIONS) or self._conforms_to(
-            ConformanceClasses.FEATURES
+        return self.conforms_to(
+            ConformanceClasses.COLLECTIONS, ConformanceClasses.FEATURES
         )
 
-    def _conforms_to(self, conformance_class: ConformanceClasses) -> bool:
-        return self._stac_io.conforms_to(conformance_class)  # type: ignore
-
     @classmethod
     def from_dict(
         cls,
@@ -236,15 +236,19 @@ def from_dict(
         return result
 
     @lru_cache()
-    def get_collection(self, collection_id: str) -> Optional[Collection]:
+    def get_collection(
+        self, collection_id: str
+    ) -> Optional[Union[Collection, CollectionClient]]:
         """Get a single collection from this Catalog/API
 
         Args:
             collection_id: The Collection ID to get
 
         Returns:
-            CollectionClient: A STAC Collection
+            Union[Collection, CollectionClient]: A STAC Collection
         """
+        collection: Union[Collection, CollectionClient]
+
         if self._supports_collections() and self._stac_io:
             url = self._get_collections_href(collection_id)
             collection = CollectionClient.from_dict(
@@ -255,21 +259,22 @@ def get_collection(self, collection_id: str) -> Optional[Collection]:
             call_modifier(self.modifier, collection)
             return collection
         else:
-            for col in self.get_collections():
-                if col.id == collection_id:
-                    call_modifier(self.modifier, col)
-                    return col
+            respond("fallback_to_pystac", "Falling back to pystac. This might be slow.")
+            for collection in super().get_collections():
+                if collection.id == collection_id:
+                    call_modifier(self.modifier, collection)
+                    return collection
 
         return None
 
-    def get_collections(self) -> Iterator[Collection]:
+    def get_collections(self) -> Iterator[Union[Collection, CollectionClient]]:
         """Get Collections in this Catalog
 
             Gets the collections from the /collections endpoint if supported,
             otherwise fall back to Catalog behavior of following child links
 
         Return:
-            Iterator[Collection]: Iterator over Collections in Catalog/API
+            Iterator[Union[Collection, CollectionClient]]: Collections in Catalog/API
         """
         collection: Union[Collection, CollectionClient]
 
@@ -285,6 +290,7 @@ def get_collections(self) -> Iterator[Collection]:
                     call_modifier(self.modifier, collection)
                     yield collection
         else:
+            respond("fallback_to_pystac", "Falling back to pystac. This might be slow.")
             for collection in super().get_collections():
                 call_modifier(self.modifier, collection)
                 yield collection
@@ -296,10 +302,11 @@ def get_items(self) -> Iterator["Item_Type"]:
             Iterator[Item]:: Iterator of items whose parent is this
                 catalog.
         """
-        if self._conforms_to(ConformanceClasses.ITEM_SEARCH):
+        if self.conforms_to(ConformanceClasses.ITEM_SEARCH):
             search = self.search()
             yield from search.items()
         else:
+            respond("fallback_to_pystac", "Falling back to pystac. This might be slow.")
             for item in super().get_items():
                 call_modifier(self.modifier, item)
                 yield item
@@ -313,13 +320,7 @@ def get_all_items(self) -> Iterator["Item_Type"]:
                 catalogs or collections connected to this catalog through
                 child links.
         """
-        if self._conforms_to(ConformanceClasses.ITEM_SEARCH):
-            # these are already modified
-            yield from self.get_items()
-        else:
-            for item in super().get_items():
-                call_modifier(self.modifier, item)
-                yield item
+        yield from self.get_items()
 
     def search(
         self,
@@ -438,26 +439,8 @@ def search(
                 or does not have a link with
                 a ``"rel"`` type of ``"search"``.
         """
-        if not self._conforms_to(ConformanceClasses.ITEM_SEARCH):
-            raise NotImplementedError(
-                "This catalog does not support search because it "
-                f'does not conform to "{ConformanceClasses.ITEM_SEARCH}"'
-            )
-        search_link = self.get_search_link()
-        if search_link:
-            if isinstance(search_link.target, str):
-                search_href = search_link.target
-            else:
-                raise NotImplementedError(
-                    "Link with rel=search was an object rather than a URI"
-                )
-        else:
-            raise NotImplementedError(
-                "No link with rel=search could be found in this catalog"
-            )
-
         return ItemSearch(
-            url=search_href,
+            url=self._get_search_href(),
             method=method,
             max_items=max_items,
             stac_io=self._stac_io,
@@ -497,35 +480,27 @@ def get_search_link(self) -> Optional[pystac.Link]:
             None,
         )
 
+    def _get_search_href(self) -> str:
+        search_link = self.get_search_link()
+        href = self._get_href("search", search_link, "search")
+        return href
+
     def _get_collections_href(self, collection_id: Optional[str] = None) -> str:
-        self_href = self.get_self_href()
-        if self_href is None:
-            data_link = self.get_single_link("data")
-            if data_link is None:
-                raise ValueError(
-                    "cannot build a collections href without a self href or a data link"
-                )
-            else:
-                collections_href = data_link.href
+        data_link = self.get_single_link("data")
+        href = self._get_href("data", data_link, "collections")
+        if collection_id is None:
+            return href
         else:
-            collections_href = f"{self_href.rstrip('/')}/collections"
-
-        if not pystac.utils.is_absolute_href(collections_href):
-            collections_href = self._make_absolute_href(collections_href)
+            return f"{href.rstrip('/')}/{collection_id}"
 
-        if collection_id is None:
-            return collections_href
+    def _get_href(self, rel: str, link: Optional[pystac.Link], endpoint: str) -> str:
+        if link and isinstance(link.href, str):
+            href = link.href
+            if not pystac.utils.is_absolute_href(href):
+                href = pystac.utils.make_absolute_href(href, self.self_href)
         else:
-            return f"{collections_href.rstrip('/')}/{collection_id}"
-
-    def _make_absolute_href(self, href: str) -> str:
-        self_link = self.get_single_link("self")
-        if self_link is None:
-            raise ValueError("cannot build an absolute href without a self link")
-        elif not pystac.utils.is_absolute_href(self_link.href):
-            raise ValueError(
-                "cannot build an absolute href from "
-                f"a relative self link: {self_link.href}"
+            respond(
+                "missing_link", f"No link with {rel=} could be found in this catalog"
             )
-        else:
-            return pystac.utils.make_absolute_href(href, self_link.href)
+            href = f"{self.self_href.rstrip('/')}/{endpoint}"
+        return href

pystac_client/item_search.py

@@ -278,7 +278,7 @@ def __init__(
         else:
             self._stac_io = StacApiIO()
 
-        self._assert_conforms_to(ConformanceClasses.ITEM_SEARCH)
+        self._stac_io.conforms_to(ConformanceClasses.ITEM_SEARCH)
 
         self._max_items = max_items
         if self._max_items is not None and limit is not None:
@@ -308,9 +308,6 @@ def __init__(
             k: v for k, v in params.items() if v is not None
         }
 
-    def _assert_conforms_to(self, conformance_class: ConformanceClasses) -> None:
-        self._stac_io.assert_conforms_to(conformance_class)
-
     def get_parameters(self) -> Dict[str, Any]:
         if self.method == "POST":
             return self._parameters
@@ -369,7 +366,7 @@ def _format_query(self, value: Optional[QueryLike]) -> Optional[Dict[str, Any]]:
         if value is None:
             return None
 
-        self._assert_conforms_to(ConformanceClasses.QUERY)
+        self._stac_io.conforms_to(ConformanceClasses.QUERY)
 
         if isinstance(value, dict):
             return value
@@ -418,7 +415,7 @@ def _format_filter(self, value: Optional[FilterLike]) -> Optional[FilterLike]:
         if value is None:
             return None
 
-        self._assert_conforms_to(ConformanceClasses.FILTER)
+        self._stac_io.conforms_to(ConformanceClasses.FILTER)
 
         return value
 
@@ -562,7 +559,7 @@ def _format_sortby(self, value: Optional[SortbyLike]) -> Optional[Sortby]:
         if value is None:
             return None
 
-        self._assert_conforms_to(ConformanceClasses.SORT)
+        self._stac_io.conforms_to(ConformanceClasses.SORT)
 
         if isinstance(value, str):
             return [self._sortby_part_to_dict(part) for part in value.split(",")]
@@ -599,7 +596,7 @@ def _format_fields(self, value: Optional[FieldsLike]) -> Optional[Fields]:
         if value is None:
             return None
 
-        self._assert_conforms_to(ConformanceClasses.FIELDS)
+        self._stac_io.conforms_to(ConformanceClasses.FIELDS)
 
         if isinstance(value, str):
             return self._fields_to_dict(value.split(","))

pystac_client/options.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+from typing import Literal, TypedDict, cast, Any
+
+
+T_Keys = Literal["on_does_not_conform_to", "on_missing_link", "on_fallback_to_pystac"]
+T_Values = Literal["ignore", "warn", "error"]
+
+
+class T_Options(TypedDict):
+    on_does_not_conform_to: T_Values
+    on_missing_link: T_Values
+    on_fallback_to_pystac: T_Values
+
+
+OPTIONS: T_Options = {
+    "on_does_not_conform_to": "warn",
+    "on_missing_link": "ignore",
+    "on_fallback_to_pystac": "ignore",
+}
+
+
+_ON_OPTIONS = frozenset(["ignore", "warn", "error"])
+
+
+_VALIDATORS = {
+    "on_does_not_conform_to": _ON_OPTIONS.__contains__,
+    "on_missing_link": _ON_OPTIONS.__contains__,
+    "on_fallback_to_pystac": _ON_OPTIONS.__contains__,
+}
+
+
+class set_options:
+    """
+    Set options for pystac-client in a controlled context.
+
+    Parameters
+    ----------
+    on_does_not_conform_to : {"ignore", "warn", "error"}, default: "warn"
+        How to inform user when client does not conform to extension
+
+        * ``ignore`` : to silently allow
+        * ``warn`` : to raise a warning
+        * ``error`` : to raise an error
+
+    on_missing_link : {"ignore", "warn", "error"}, default: "ignore"
+        How to inform user when link is properly implemented
+
+        * ``ignore`` : to silently allow
+        * ``warn`` : to raise a warning
+        * ``error`` : to raise an error
+
+    on_fallback_to_pystac : {"ignore", "warn", "error"}, default: "ignore"
+        How to inform user when falling back to pystac implementation
+
+        * ``ignore`` : to silently allow
+        * ``warn`` : to raise a warning
+        * ``error`` : to raise an error
+
+    Examples
+    --------
+    It is possible to use ``set_options`` either as a context manager:
+
+    >>> with set_options(on_does_not_conform_to="error"):
+    ...     Client.open(url)
+
+    Or to set global options:
+
+    >>> set_options(on_fallback_to_pystac="warn")
+    """
+
+    def __init__(self, **kwargs: T_Values):
+        self.old: T_Options = OPTIONS.copy()
+        for k, v in kwargs.items():
+            if k not in OPTIONS:
+                raise ValueError(
+                    f"argument name {k!r} is not in the set of valid "
+                    "options {set(OPTIONS)!r}"
+                )
+            if k in _VALIDATORS and not _VALIDATORS[k](v):
+                expected = f"Expected one of {_ON_OPTIONS!r}"
+                raise ValueError(
+                    f"option {k!r} given an invalid value: {v!r}." + expected
+                )
+        self._apply_update(**kwargs)
+
+    def _apply_update(self, **kwargs: T_Values) -> None:
+        OPTIONS.update(kwargs)  # type: ignore
+
+    def __enter__(self) -> None:
+        return
+
+    def __exit__(self, *args: Any) -> None:
+        self._apply_update(**cast(T_Options, self.old))
+
+
+def get_options() -> T_Options:
+    """
+    Get options for pystac-client
+
+    See Also
+    ----------
+    set_options
+
+    """
+    return OPTIONS