stac-utils · duckontheweb · Feb 4, 2022 · Feb 1, 2022 · Feb 1, 2022 · Feb 1, 2022
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6  5,7 @@
 ### Added
 
 - Experimental support for Python 3.11 ([#731](https://github.com/stac-utils/pystac/pull/731))
 - Accept PathLike objects in `StacIO` I/O methods, `pystac.read_file` and `pystac.write_file` ([#728](https://github.com/stac-utils/pystac/pull/728))
 
 ### Removed
 

diff --git a/pystac/__init__.py b/pystac/__init__.py
@@ -39,6  39,9 @@
     "set_stac_version",
 ]
 
 import os
 from typing import Any, AnyStr, Dict, Optional, Union
 
 from pystac.errors import (
     STACError,
     STACTypeError,
@@ -50,7  53,6 @@
     STACValidationError,
 )
 
-from typing import Any, Dict, Optional
 from pystac.version import (
     __version__,
     get_stac_version,
@@ -60,7  62,7 @@
 from pystac.rel_type import RelType
 from pystac.stac_io import StacIO
 from pystac.stac_object import STACObject, STACObjectType
-from pystac.link import Link, HIERARCHICAL_LINKS
 from pystac.link import Link, HIERARCHICAL_LINKS, HREF
 from pystac.catalog import Catalog, CatalogType
 from pystac.collection import (
     Collection,
@@ -112,7  114,7 @@
 )
 
 
-def read_file(href: str, stac_io: Optional[StacIO] = None) -> STACObject:
 def read_file(href: HREF, stac_io: Optional[StacIO] = None) -> STACObject:
     """Reads a STAC object from a file.
 
     This method will return either a Catalog, a Collection, or an Item based on what
@@ -144,7  146,7 @@ def read_file(href: str, stac_io: Optional[StacIO] = None) -> STACObject:
 def write_file(
     obj: STACObject,
     include_self_link: bool = True,
-    dest_href: Optional[str] = None,
     dest_href: Optional[HREF] = None,
     stac_io: Optional[StacIO] = None,
 ) -> None:
     """Writes a STACObject to a file.
@@ -170,6  172,7 @@ def write_file(
     """
     if stac_io is None:
         stac_io = StacIO.default()
     dest_href = None if dest_href is None else str(os.fspath(dest_href))
     obj.save_object(
         include_self_link=include_self_link, dest_href=dest_href, stac_io=stac_io
     )

diff --git a/pystac/link.py b/pystac/link.py
@@ -1,3  1,4 @@
 import os
 from copy import copy
 from typing import Any, Dict, Optional, TYPE_CHECKING, Union
 
@@ -10,6  11,13 @@
     from pystac.catalog import Catalog as Catalog_Type
     from pystac.collection import Collection as Collection_Type
 
     PathLike = os.PathLike[str]
 
 else:
     PathLike = os.PathLike
 
 HREF = Union[str, os.PathLike]
 
 HIERARCHICAL_LINKS = [
     pystac.RelType.ROOT,
     pystac.RelType.CHILD,
@@ -20,7  28,7 @@
 ]
 
 
-class Link:
 class Link(PathLike):
     """A link connects a :class:`~pystac.STACObject` to another entity.
 
     The target of a link can be either another STACObject, or
@@ -240,6  248,9 @@ def has_target_href(self) -> bool:
         """Returns true if this link has a string href in its target information."""
         return self._target_href is not None
 
     def __fspath__(self) -> str:
         return self.absolute_href
 
     def __repr__(self) -> str:
         return "<Link rel={} target={}>".format(self.rel, self.target)
 
@@ -407,9  418,10 @@ def collection(cls, c: "Collection_Type") -> "Link":
         return cls(pystac.RelType.COLLECTION, c, media_type=pystac.MediaType.JSON)
 
     @classmethod
-    def self_href(cls, href: str) -> "Link":
     def self_href(cls, href: HREF) -> "Link":
         """Creates a self link to a file's location."""
-        return cls(pystac.RelType.SELF, href, media_type=pystac.MediaType.JSON)
         href_str = str(os.fspath(href))
         return cls(pystac.RelType.SELF, href_str, media_type=pystac.MediaType.JSON)
 
     @classmethod
     def child(cls, c: "Catalog_Type", title: Optional[str] = None) -> "Link":

diff --git a/pystac/stac_io.py b/pystac/stac_io.py
@@ -9,13  9,13 @@
     Optional,
     TYPE_CHECKING,
     Tuple,
-    Union,
 )
 
 from urllib.request import urlopen
 from urllib.error import HTTPError
 
 import pystac
 from pystac.link import HREF
 from pystac.utils import safe_urlparse
 from pystac.serialization import (
     merge_common_properties,
@@ -33,16  33,13 @@
 if TYPE_CHECKING:
     from pystac.stac_object import STACObject as STACObject_Type
     from pystac.catalog import Catalog as Catalog_Type
-    from pystac.link import Link as Link_Type
 
 
 class StacIO(ABC):
     _default_io: Optional[Callable[[], "StacIO"]] = None
 
     @abstractmethod
-    def read_text(
-        self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any
-    ) -> str:
     def read_text(self, source: HREF, *args: Any, **kwargs: Any) -> str:
         """Read text from the given URI.
 
         The source to read from can be specified as a string or a
@@ -66,7  63,11 @@ def read_text(
 
     @abstractmethod
     def write_text(
-        self, dest: Union[str, "Link_Type"], txt: str, *args: Any, **kwargs: Any
         self,
         dest: HREF,
         txt: str,
         *args: Any,
         **kwargs: Any,
     ) -> None:
         """Write the given text to a file at the given URI.
 
@@ -124,7  125,7 @@ def json_dumps(self, json_dict: Dict[str, Any], *args: Any, **kwargs: Any) -> st
     def stac_object_from_dict(
         self,
         d: Dict[str, Any],
-        href: Optional[str] = None,
         href: Optional[HREF] = None,
         root: Optional["Catalog_Type"] = None,
         preserve_dict: bool = True,
     ) -> "STACObject_Type":
@@ -143,41  144,40 @@ def stac_object_from_dict(
                 parameter. Set to ``False`` when possible to avoid the performance
                 hit of a deepcopy.
         """
         href_str = None if href is None else str(os.fspath(href))
         if identify_stac_object_type(d) == pystac.STACObjectType.ITEM:
             collection_cache = None
             if root is not None:
                 collection_cache = root._resolved_objects.as_collection_cache()
 
             # Merge common properties in case this is an older STAC object.
             merge_common_properties(
-                d, json_href=href, collection_cache=collection_cache
                 d, json_href=href_str, collection_cache=collection_cache
             )
 
         info = identify_stac_object(d)
         d = migrate_to_latest(d, info)
 
         if info.object_type == pystac.STACObjectType.CATALOG:
             result = pystac.Catalog.from_dict(
-                d, href=href, root=root, migrate=False, preserve_dict=preserve_dict
                 d, href=href_str, root=root, migrate=False, preserve_dict=preserve_dict
             )
             result._stac_io = self
             return result
 
         if info.object_type == pystac.STACObjectType.COLLECTION:
             return pystac.Collection.from_dict(
-                d, href=href, root=root, migrate=False, preserve_dict=preserve_dict
                 d, href=href_str, root=root, migrate=False, preserve_dict=preserve_dict
             )
 
         if info.object_type == pystac.STACObjectType.ITEM:
             return pystac.Item.from_dict(
-                d, href=href, root=root, migrate=False, preserve_dict=preserve_dict
                 d, href=href_str, root=root, migrate=False, preserve_dict=preserve_dict
             )
 
         raise ValueError(f"Unknown STAC object type {info.object_type}")
 
-    def read_json(
-        self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any
-    ) -> Dict[str, Any]:
     def read_json(self, source: HREF, *args: Any, **kwargs: Any) -> Dict[str, Any]:
         """Read a dict from the given source.
 
         See :func:`StacIO.read_text <pystac.StacIO.read_text>` for usage of
@@ -199,7  199,7 @@ def read_json(
 
     def read_stac_object(
         self,
-        source: Union[str, "Link_Type"],
         source: HREF,
         root: Optional["Catalog_Type"] = None,
         *args: Any,
         **kwargs: Any,
@@ -224,12  224,13 @@ def read_stac_object(
             contained in the file at the given uri.
         """
         d = self.read_json(source, *args, **kwargs)
-        href = source if isinstance(source, str) else source.get_absolute_href()
-        return self.stac_object_from_dict(d, href=href, root=root, preserve_dict=False)
         return self.stac_object_from_dict(
             d, href=source, root=root, preserve_dict=False
         )
 
     def save_json(
         self,
-        dest: Union[str, "Link_Type"],
         dest: HREF,
         json_dict: Dict[str, Any],
         *args: Any,
         **kwargs: Any,
@@ -264,18  265,12 @@ def default(cls) -> "StacIO":
 
 
 class DefaultStacIO(StacIO):
-    def read_text(self, source: Union[str, "Link_Type"], *_: Any, **__: Any) -> str:
     def read_text(self, source: HREF, *_: Any, **__: Any) -> str:
         """A concrete implementation of :meth:`StacIO.read_text
         <pystac.StacIO.read_text>`. Converts the ``source`` argument to a string (if it
         is not already) and delegates to :meth:`DefaultStacIO.read_text_from_href` for
         opening and reading the file."""
-        href: Optional[str]
-        if isinstance(source, str):
-            href = source
-        else:
-            href = source.get_absolute_href()
-            if href is None:
-                raise IOError(f"Could not get an absolute HREF from link {source}")
         href = str(os.fspath(source))
         return self.read_text_from_href(href)
 
     def read_text_from_href(self, href: str) -> str:
@@ -302,20  297,12 @@ def read_text_from_href(self, href: str) -> str:
                 href_contents = f.read()
         return href_contents
 
-    def write_text(
-        self, dest: Union[str, "Link_Type"], txt: str, *_: Any, **__: Any
-    ) -> None:
     def write_text(self, dest: HREF, txt: str, *_: Any, **__: Any) -> None:
         """A concrete implementation of :meth:`StacIO.write_text
         <pystac.StacIO.write_text>`. Converts the ``dest`` argument to a string (if it
         is not already) and delegates to :meth:`DefaultStacIO.write_text_from_href` for
         opening and reading the file."""
-        href: Optional[str]
-        if isinstance(dest, str):
-            href = dest
-        else:
-            href = dest.get_absolute_href()
-            if href is None:
-                raise IOError(f"Could not get an absolute HREF from link {dest}")
         href = str(os.fspath(dest))
         return self.write_text_to_href(href, txt)
 
     def write_text_to_href(self, href: str, txt: str) -> None:
@@ -329,6  316,7 @@ def write_text_to_href(self, href: str, txt: str) -> None:
             href : The path to which the file will be written.
             txt : The string content to write to the file.
         """
         href = os.fspath(href)
         dirname = os.path.dirname(href)
         if dirname != "" and not os.path.isdir(dirname):
             os.makedirs(dirname)
@@ -357,9  345,7 @@ def json_loads(self, txt: str, *_: Any, **__: Any) -> Dict[str, Any]:
         )
         return result
 
-    def read_json(
-        self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any
-    ) -> Dict[str, Any]:
     def read_json(self, source: HREF, *args: Any, **kwargs: Any) -> Dict[str, Any]:
         """Overwrites :meth:`StacIO.read_json <pystac.StacIO.read_json>` for
         deserializing a JSON file to a dictionary while checking for duplicate object
         keys.
@@ -372,7  358,7 @@ def read_json(
         try:
             return self.json_loads(txt, source=source)
         except pystac.DuplicateObjectKeyError as e:
-            url = source if isinstance(source, str) else source.get_absolute_href()
             url = str(os.fspath(source))
             msg = str(e)   f" in {url}"
             raise pystac.DuplicateObjectKeyError(msg)
 

diff --git a/tests/test_link.py b/tests/test_link.py
@@ -22,6  22,13 @@ def setUp(self) -> None:
             properties={},
         )
 
     def test_path_like(self) -> None:
         rel = "some-rel"
         target = os.path.abspath("../elsewhere")
         link = pystac.Link(rel, target)
 
         self.assertEqual(os.fspath(link), target)
 
     def test_minimal(self) -> None:
         rel = "my rel"
         target = "https://example.com/a/b"

diff --git a/tests/utils/stac_io_mock.py b/tests/utils/stac_io_mock.py
@@ -1,8  1,10 @@
-from typing import Any, Optional, Union
 import os
 from typing import Any, AnyStr, Optional, Union
 from unittest.mock import Mock
 
 import pystac
 from pystac.stac_io import DefaultStacIO, StacIO
 from pystac.link import HREF
 
 
 class MockStacIO(pystac.StacIO):
@@ -24,14  26,16 @@ def __init__(self, wrapped_stac_io: Optional[StacIO] = None) -> None:
         else:
             self.wrapped_stac_io = wrapped_stac_io
 
-    def read_text(
-        self, source: Union[str, pystac.Link], *args: Any, **kwargs: Any
-    ) -> str:
     def read_text(self, source: HREF, *args: Any, **kwargs: Any) -> str:
         self.mock.read_text(source)
         return self.wrapped_stac_io.read_text(source)
 
     def write_text(
-        self, dest: Union[str, pystac.Link], txt: str, *args: Any, **kwargs: Any
         self,
         dest: Union[str, "os.PathLike[AnyStr]"],
         txt: str,
         *args: Any,
         **kwargs: Any
     ) -> None:
         self.mock.write_text(dest, txt)
         self.wrapped_stac_io.write_text(dest, txt)