typing improvements

.github/pull_request_template.md

@@ -1,4 +1,6 @@
-**Related Issue(s):** #
+**Related Issue(s):** 
+
+- #
 
 
 **Description:**

pystac_client/client.py

@@ -3,6 +3,7 @@
 
 import pystac
 import pystac.validation
+from pystac import Collection
 
 from pystac_client.collection_client import CollectionClient
 from pystac_client.conformance import ConformanceClasses
@@ -60,25 +61,30 @@ def open(
         Return:
             catalog : A :class:`Client` instance for this Catalog/API
         """
-        cat = cls.from_file(url, headers=headers, parameters=parameters)
-        search_link = cat.get_search_link()
+        client: Client = cls.from_file(url, headers=headers, parameters=parameters)
+        search_link = client.get_search_link()
         # if there is a search link, but no conformsTo advertised, ignore
         # conformance entirely
         # NOTE: this behavior to be deprecated as implementations become conformant
-        if ignore_conformance or (
-            "conformsTo" not in cat.extra_fields.keys()
-            and search_link
-            and search_link.href
-            and len(search_link.href) > 0
+        if client._stac_io and (
+            ignore_conformance
+            or (
+                client
+                and "conformsTo" not in client.extra_fields.keys()
+                and search_link
+                and search_link.href
+                and len(search_link.href) > 0
+            )
         ):
-            cat._stac_io.set_conformance(None)
-        return cat
+            client._stac_io.set_conformance(None)  # type: ignore
+
+        return client
 
     @classmethod
-    def from_file(
+    def from_file(  # type: ignore
         cls,
         href: str,
-        stac_io: Optional[pystac.StacIO] = None,
+        stac_io: Optional[StacApiIO] = None,
         headers: Optional[Dict[str, str]] = None,
         parameters: Optional[Dict[str, Any]] = None,
     ) -> "Client":
@@ -90,20 +96,21 @@ def from_file(
         if stac_io is None:
             stac_io = StacApiIO(headers=headers, parameters=parameters)
 
-        cat = super().from_file(href, stac_io)
+        client: Client = super().from_file(href, stac_io)  # type: ignore
 
-        cat._stac_io._conformance = cat.extra_fields.get("conformsTo", [])
+        client._stac_io._conformance = client.extra_fields.get(  # type: ignore
+            "conformsTo", []
+        )
 
-        return cat
+        return client
 
     def _supports_collections(self) -> bool:
         return self._conforms_to(ConformanceClasses.COLLECTIONS) or self._conforms_to(
             ConformanceClasses.FEATURES
         )
 
-    # TODO: fix this with the stac_api_io() method in a future PR
     def _conforms_to(self, conformance_class: ConformanceClasses) -> bool:
-        return self._stac_io.conforms_to(conformance_class)  # type: ignore
+        return self._stac_io.conforms_to(conformance_class)
 
     @classmethod
     def from_dict(
@@ -125,7 +132,7 @@ def from_dict(
             )
 
     @lru_cache()
-    def get_collection(self, collection_id: str) -> CollectionClient:
+    def get_collection(self, collection_id: str) -> Optional[Collection]:
         """Get a single collection from this Catalog/API
 
         Args:
@@ -134,7 +141,7 @@ def get_collection(self, collection_id: str) -> CollectionClient:
         Returns:
             CollectionClient: A STAC Collection
         """
-        if self._supports_collections():
+        if self._supports_collections() and self._stac_io:
             url = f"{self.get_self_href()}/collections/{collection_id}"
             collection = CollectionClient.from_dict(
                 self._stac_io.read_json(url), root=self
@@ -145,7 +152,9 @@ def get_collection(self, collection_id: str) -> CollectionClient:
                 if col.id == collection_id:
                     return col
 
-    def get_collections(self) -> Iterable[CollectionClient]:
+        return None
+
+    def get_collections(self) -> Iterable[Collection]:
         """Get Collections in this Catalog
 
             Gets the collections from the /collections endpoint if supported,
@@ -154,9 +163,9 @@ def get_collections(self) -> Iterable[CollectionClient]:
         Return:
             Iterable[CollectionClient]: Iterator through Collections in Catalog/API
         """
-        if self._supports_collections():
-            url = self.get_self_href() + "/collections"
-            for page in self._stac_io.get_pages(url):
+        if self._supports_collections() and self.get_self_href() is not None:
+            url = f"{self.get_self_href()}/collections"
+            for page in self._stac_io.get_pages(url):  # type: ignore
                 if "collections" not in page:
                     raise APIError("Invalid response from /collections")
                 for col in page["collections"]:
@@ -228,14 +237,21 @@ def search(self, **kwargs: Any) -> ItemSearch:
                 f'does not conform to "{ConformanceClasses.ITEM_SEARCH}"'
             )
         search_link = self.get_search_link()
-        if search_link is None:
+        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" type of "search" could be found in this catalog'
+                "No link with rel=search could be found in this catalog"
             )
 
         return ItemSearch(
-            search_link.target,
-            stac_io=self._stac_io,
+            url=search_href,
+            stac_io=self._stac_io,  # type: ignore
             client=self,
             **kwargs,
         )

pystac_client/collection_client.py

@@ -29,7 +29,9 @@ def get_items(self) -> Iterable["Item_Type"]:
         link = self.get_single_link("items")
         root = self.get_root()
         if link is not None and root is not None:
-            search = ItemSearch(link.href, method="GET", stac_io=root._stac_io)
+            search = ItemSearch(
+                url=link.href, method="GET", stac_io=root._stac_io
+            )  # type: ignore
             yield from search.items()
         else:
             yield from super().get_items()

pystac_client/item_search.py

@@ -12,13 +12,12 @@
 from dateutil.relativedelta import relativedelta
 from dateutil.tz import tzutc
 from pystac import Collection, Item, ItemCollection
-from pystac.stac_io import StacIO
 
 from pystac_client.conformance import ConformanceClasses
 from pystac_client.stac_api_io import StacApiIO
 
 if TYPE_CHECKING:
-    from pystac_client.client import Client
+    from pystac_client import client
 
 DATETIME_REGEX = re.compile(
     r"(?P<year>\d{4})(\-(?P<month>\d{2})(\-(?P<day>\d{2})"
@@ -219,8 +218,8 @@ def __init__(
         *,
         method: Optional[str] = "POST",
         max_items: Optional[int] = DEFAULT_LIMIT_AND_MAX_ITEMS,
-        stac_io: Optional[StacIO] = None,
-        client: Optional["Client"] = None,
+        stac_io: Optional[StacApiIO] = None,
+        client: Optional["client.Client"] = None,
         limit: Optional[int] = DEFAULT_LIMIT_AND_MAX_ITEMS,
         ids: Optional[IDsLike] = None,
         collections: Optional[CollectionsLike] = None,
@@ -268,9 +267,8 @@ def __init__(
 
         self._parameters = {k: v for k, v in params.items() if v is not None}
 
-    # TODO: fix this with the stac_api_io() method in a future PR
     def _assert_conforms_to(self, conformance_class: ConformanceClasses) -> None:
-        self._stac_io.assert_conforms_to(conformance_class)  # type: ignore
+        self._stac_io.assert_conforms_to(conformance_class)
 
     def get_parameters(self) -> Dict[str, Any]:
         if self.method == "POST":
@@ -293,7 +291,7 @@ def get_parameters(self) -> Dict[str, Any]:
         else:
             raise Exception(f"Unsupported method {self.method}")
 
-    def _format_query(self, value: QueryLike) -> Optional[Dict[str, Any]]:
+    def _format_query(self, value: Optional[QueryLike]) -> Optional[Dict[str, Any]]:
         if value is None:
             return None
 
@@ -364,12 +362,14 @@ def _format_bbox(value: Optional[BBoxLike]) -> Optional[BBox]:
 
     @staticmethod
     def _format_datetime(value: Optional[DatetimeLike]) -> Optional[Datetime]:
-        def _to_utc_isoformat(dt):
+        def _to_utc_isoformat(dt: datetime_) -> str:
             dt = dt.astimezone(timezone.utc)
             dt = dt.replace(tzinfo=None)
-            return dt.isoformat("T") + "Z"
+            return f'{dt.isoformat("T")}Z'
 
-        def _to_isoformat_range(component: DatetimeOrTimestamp):
+        def _to_isoformat_range(
+            component: DatetimeOrTimestamp,
+        ) -> Tuple[Optional[str], Optional[str]]:
             """Converts a single DatetimeOrTimestamp into one or two Datetimes.
 
             This is required to expand a single value like "2017" out to the whole
@@ -452,20 +452,20 @@ def _to_isoformat_range(component: DatetimeOrTimestamp):
 
     @staticmethod
     def _format_collections(value: Optional[CollectionsLike]) -> Optional[Collections]:
-        def _format(c: Any) -> Any:
+        def _format(c: Any) -> Collections:
             if isinstance(c, str):
-                return c
+                return (c,)
             if isinstance(c, Iterable):
-                return tuple(map(_format, c))
+                return tuple(map(lambda x: _format(x)[0], c))
 
-            return c.id
+            return (c.id,)
 
         if value is None:
             return None
         if isinstance(value, str):
-            return tuple(map(_format, value.split(",")))
+            return tuple(map(lambda x: _format(x)[0], value.split(",")))
         if isinstance(value, Collection):
-            return (_format(value),)
+            return _format(value)
 
         return _format(value)
 
@@ -490,7 +490,7 @@ def _format_sortby(self, value: Optional[SortbyLike]) -> Optional[Sortby]:
 
         if isinstance(value, list):
             if value and isinstance(value[0], str):
-                return [self._sortby_part_to_dict(v) for v in value]
+                return [self._sortby_part_to_dict(str(v)) for v in value]
             elif value and isinstance(value[0], dict):
                 return value
 
@@ -559,9 +559,9 @@ def _format_intersects(value: Optional[IntersectsLike]) -> Optional[Intersects]:
         if isinstance(value, dict):
             return deepcopy(value)
         if isinstance(value, str):
-            return json.loads(value)
+            return dict(json.loads(value))
         if hasattr(value, "__geo_interface__"):
-            return deepcopy(getattr(value, "__geo_interface__"))
+            return dict(deepcopy(getattr(value, "__geo_interface__")))
         raise Exception(
             "intersects must be of type None, str, dict, or an object that "
             "implements __geo_interface__"
@@ -610,10 +610,13 @@ def item_collections(self) -> Iterator[ItemCollection]:
             ItemCollection : a group of Items matching the search criteria within an
             ItemCollection
         """
-        for page in self._stac_io.get_pages(
-            self.url, self.method, self.get_parameters()
-        ):
-            yield ItemCollection.from_dict(page, preserve_dict=False, root=self.client)
+        if isinstance(self._stac_io, StacApiIO):
+            for page in self._stac_io.get_pages(
+                self.url, self.method, self.get_parameters()
+            ):
+                yield ItemCollection.from_dict(
+                    page, preserve_dict=False, root=self.client
+                )
 
     def get_items(self) -> Iterator[Item]:
         """DEPRECATED. Use :meth:`ItemSearch.items` instead.

pystac_client/stac_api_io.py

@@ -58,7 +58,7 @@ def read_text(
         self,
         source: Union[str, Link],
         *args: Any,
-        parameters: Optional[dict] = {},
+        parameters: Optional[Dict[str, Any]] = None,
         **kwargs: Any,
     ) -> str:
         """Read text from the given URI.
@@ -104,20 +104,20 @@ def read_text(
     def request(
         self,
         href: str,
-        method: Optional[str] = "GET",
-        headers: Optional[dict] = {},
-        parameters: Optional[dict] = {},
+        method: Optional[str] = None,
+        headers: Optional[Dict[str, str]] = None,
+        parameters: Optional[Dict[str, Any]] = None,
     ) -> str:
         """Makes a request to an http endpoint
 
         Args:
             href (str): The request URL
             method (Optional[str], optional): The http method to use, 'GET' or 'POST'.
-              Defaults to 'GET'.
-            headers (Optional[dict], optional): Additional headers to include in
-              request. Defaults to {}.
-            parameters (Optional[dict], optional): parameters to send with request.
-              Defaults to {}.
+              Defaults to None, which will result in 'GET' being used.
+            headers (Optional[Dict[str, str]], optional): Additional headers to include
+                in request. Defaults to None.
+            parameters (Optional[Dict[str, Any]], optional): parameters to send with
+                request. Defaults to None.
 
         Raises:
             APIError: raised if the server returns an error response
@@ -128,10 +128,10 @@ def request(
         if method == "POST":
             request = Request(method=method, url=href, headers=headers, json=parameters)
         else:
-            params = deepcopy(parameters)
+            params = deepcopy(parameters) or {}
             if "intersects" in params:
                 params["intersects"] = json.dumps(params["intersects"])
-            request = Request(method=method, url=href, headers=headers, params=params)
+            request = Request(method="GET", url=href, headers=headers, params=params)
         try:
             prepped = self.session.prepare_request(request)
             msg = f"{prepped.method} {prepped.url} Headers: {prepped.headers}"
@@ -188,14 +188,14 @@ def stac_object_from_dict(
         d = migrate_to_latest(d, info)
 
         if info.object_type == pystac.STACObjectType.CATALOG:
-            result = pystac_client.Client.from_dict(
+            result = pystac_client.client.Client.from_dict(
                 d, href=href, root=root, migrate=False, preserve_dict=preserve_dict
             )
             result._stac_io = self
             return result
 
         if info.object_type == pystac.STACObjectType.COLLECTION:
-            return pystac_client.CollectionClient.from_dict(
+            return pystac_client.collection_client.CollectionClient.from_dict(
                 d, href=href, root=root, migrate=False, preserve_dict=preserve_dict
             )
 
@@ -206,7 +206,12 @@ def stac_object_from_dict(
 
         raise ValueError(f"Unknown STAC object type {info.object_type}")
 
-    def get_pages(self, url, method="GET", parameters={}) -> Iterator[Dict]:
+    def get_pages(
+        self,
+        url: str,
+        method: Optional[str] = None,
+        parameters: Optional[Dict[str, Any]] = None,
+    ) -> Iterator[Dict[str, Any]]:
         """Iterator that yields dictionaries for each page at a STAC paging
         endpoint, e.g., /collections, /search
 
@@ -273,5 +278,5 @@ def conforms_to(self, conformance_class: ConformanceClasses) -> bool:
         return True
 
     def set_conformance(self, conformance: Optional[List[str]]) -> None:
-        """Sets (or clears) the conformances for this StacIO."""
+        """Sets (or clears) the conformance classes for this StacIO."""
         self._conformance = conformance

tests/test_client.py

@@ -355,7 +355,9 @@ def test_no_search_link(self, api):
 
         with pytest.raises(NotImplementedError) as excinfo:
             api.search(limit=10, max_items=10, collections="naip")
-        assert 'No link with "rel" type of "search"' in str(excinfo.value)
+        assert "No link with rel=search could be found in this catalog" in str(
+            excinfo.value
+        )
 
     def test_no_conforms_to(self) -> None:
         with open(str(TEST_DATA / "planetary-computer-root.json")) as f: