Refactor STAC_IO to StacIO

Author: lossyrob

pystac/__init__.py

@@ -7,7 +7,7 @@
 
 from typing import Any, Dict, Optional
 from pystac.version import (__version__, get_stac_version, set_stac_version)  # type:ignore
-from pystac.stac_io import STAC_IO  # type:ignore
+from pystac.stac_io import StacIO  # type:ignore
 from pystac.stac_object import (STACObject, STACObjectType)  # type:ignore
 from pystac.media_type import MediaType  # type:ignore
 from pystac.link import (Link, HIERARCHICAL_LINKS)  # type:ignore
@@ -99,7 +99,8 @@ def write_file(obj: STACObject,
 
 def read_dict(d: Dict[str, Any],
               href: Optional[str] = None,
-              root: Optional[Catalog] = None) -> STACObject:
+              root: Optional[Catalog] = None,
+              stac_io: Optional[StacIO] = None) -> STACObject:
     """Reads a STAC object from a dict representing the serialized JSON version of the
     STAC object.
 
@@ -115,5 +116,9 @@ def read_dict(d: Dict[str, Any],
         root (Catalog or Collection): Optional root of the catalog for this object.
             If provided, the root's resolved object cache can be used to search for
             previously resolved instances of the STAC object.
+        stac_io: Optional StacIO instance to use for reading. If None, the
+            default instance will be used.
     """
-    return STAC_IO.stac_object_from_dict(d, href, root)
+    if stac_io is None:
+        stac_io = StacIO.default()
+    return stac_io.stac_object_from_dict(d, href, root)

pystac/catalog.py

@@ -110,6 +110,12 @@ class Catalog(STACObject):
 
     STAC_OBJECT_TYPE = ps.STACObjectType.CATALOG
 
+    _stac_io: Optional[ps.StacIO] = None
+    """Optional instance of StacIO that will be used by default
+    for any IO operations on objects contained by this catalog.
+    Set while reading in a catalog. This is set when a catalog
+    is read by a StacIO instance."""
+
     DEFAULT_FILE_NAME = "catalog.json"
     """Default file name that will be given to this STAC object in a canonical format."""
     def __init__(self,

pystac/link.py

@@ -2,7 +2,6 @@
 from typing import Any, Dict, Optional, TYPE_CHECKING, Union, cast
 
 import pystac as ps
-from pystac.stac_io import STAC_IO
 from pystac.utils import (make_absolute_href, make_relative_href, is_absolute_href)
 
 if TYPE_CHECKING:
@@ -176,11 +175,22 @@ def resolve_stac_object(self, root: Optional["Catalog_Type"] = None) -> "Link":
                 target_href = make_absolute_href(target_href, start_href)
             obj = None
 
+            stac_io: Optional[ps.StacIO] = None
+
             if root is not None:
                 obj = root._resolved_objects.get_by_href(target_href)
+                stac_io = root._stac_io
 
             if obj is None:
-                obj = STAC_IO.read_stac_object(target_href, root=root)
+
+                if stac_io is None:
+                    if self.owner is not None:
+                        if isinstance(self.owner, ps.Catalog):
+                            stac_io = self.owner._stac_io
+                    if stac_io is None:
+                        stac_io = ps.StacIO.default()
+
+                obj = stac_io.read_stac_object(target_href, root=root)
                 obj.set_self_href(target_href)
                 if root is not None:
                     obj = root._resolved_objects.get_or_cache(obj)

pystac/serialization/__init__.py

@@ -27,8 +27,7 @@ def stac_object_from_dict(d: Dict[str, Any],
             If provided, the root's resolved object cache can be used to search for
             previously resolved instances of the STAC object.
 
-    Note: This is used internally in STAC_IO to deserialize STAC Objects.
-    It is in the top level __init__ in order to avoid circular dependencies.
+    Note: This is used internally in StacIO instances to deserialize STAC Objects.
     """
     if identify_stac_object_type(d) == ps.STACObjectType.ITEM:
         collection_cache = None

pystac/serialization/common_properties.py

@@ -11,6 +11,8 @@ def merge_common_properties(item_dict: Dict[str, Any],
                             json_href: Optional[str] = None) -> bool:
     """Merges Collection properties into an Item.
 
+    Note: This is only applicable to reading old STAC versions (pre 1.0.0-beta.1).
+
     Args:
         item_dict (dict): JSON dict of the Item which properties should be merged
             into.
@@ -70,7 +72,7 @@ def merge_common_properties(item_dict: Dict[str, Any],
                     collection = collection_cache.get_by_href(collection_href)
 
                 if collection is None:
-                    collection = ps.STAC_IO.read_json(collection_href)
+                    collection = ps.StacIO.default().read_json(collection_href)
 
     if collection is not None:
         collection_id = None

pystac/stac_io.py

@@ -1,61 +1,231 @@
+from abc import ABC, abstractmethod
 import os
 import json
-from typing import Any, Callable, Dict, Optional, TYPE_CHECKING
+from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union
 
 from urllib.parse import urlparse
 from urllib.request import urlopen
 from urllib.error import HTTPError
 
+import pystac as ps
 import pystac.serialization
 
+# Use orjson if available
+try:
+    import orjson
+except ImportError:
+    orjson = None
+
 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 STAC_IO:
-    """Methods used to read and save STAC json.
-    Allows users of the library to set their own methods
-    (e.g. for reading and writing from cloud storage)
-    """
-    @staticmethod
-    def default_read_text_method(uri: str) -> str:
-        """Default method for reading text. Only handles local file paths."""
-        parsed = urlparse(uri)
+class StacIO(ABC):
+    _default_io: Optional[Type["StacIO"]] = None
+
+    @abstractmethod
+    def read_text(self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any) -> str:
+        """Read text from the given URI.
+
+        The source to read from can be specified
+        as a string or a Link. If it's a string, it's the URL of the HREF from which to
+        read. When reading links, PySTAC will pass in the entire link body.
+        This enables implementations to utilize additional link information,
+        e.g. the "post" information in a pagination link from a STAC API search.
+
+        Args:
+            source (str or pystac.Link): The source to read from.
+
+        Returns:
+            str: The text contained in the file at the location specified by the uri.
+        """
+        raise NotImplementedError("read_text not implemented")
+
+    @abstractmethod
+    def write_text(self, dest: Union[str, "Link_Type"], txt: str, *args: Any,
+                   **kwargs: Any) -> None:
+        """Write the given text to a file at the given URI.
+
+        The destination to write to from can be specified
+        as a string or a Link. If it's a string, it's the URL of the HREF from which to
+        read. When writing based on links links, PySTAC will pass in the entire link body.
+
+        Args:
+            dest (str or pystac.Link): The destination to write to.
+            txt (str): The text to write.
+        """
+        raise NotImplementedError("write_text not implemented")
+
+    def _json_loads(self, txt: str, source: Union[str, "Link_Type"]) -> Dict[str, Any]:
+        if orjson is not None:
+            return orjson.loads(txt)
+        else:
+            return json.loads(self.read_text(txt))
+
+    def _json_dumps(self, json_dict: Dict[str, Any], source: Union[str, "Link_Type"]) -> str:
+        if orjson is not None:
+            return orjson.dumps(json_dict, option=orjson.OPT_INDENT_2).decode('utf-8')
+        else:
+            return json.dumps(json_dict, indent=2)
+
+    def stac_object_from_dict(self,
+                              d: Dict[str, Any],
+                              href: Optional[str] = None,
+                              root: Optional["Catalog_Type"] = None) -> "STACObject_Type":
+        result = pystac.serialization.stac_object_from_dict(d, href, root)
+        if isinstance(result, ps.Catalog):
+            # Set the stac_io instance for usage by io operations
+            # where this catalog is the root.
+            result._stac_io = self
+        return result
+
+    def read_json(self, source: Union[str, "Link_Type"]) -> Dict[str, Any]:
+        """Read a dict from the given source.
+
+        See :func:`StacIO.read_text <pystac.StacIO.read_text>` for usage of
+        str vs Link as a parameter.
+
+        Args:
+            source (str or Link): The source from which to read.
+
+        Returns:
+            dict: A dict representation of the JSON contained in the file at the
+            given source.
+        """
+        txt = self.read_text(source)
+        return self._json_loads(txt, source)
+
+    def read_stac_object(self,
+                         source: Union[str, "Link_Type"],
+                         root: Optional["Catalog_Type"] = None) -> "STACObject_Type":
+        """Read a STACObject from a JSON file at the given source.
+
+        See :func:`StacIO.read_text <pystac.StacIO.read_text>` for usage of
+        str vs Link as a parameter.
+
+        Args:
+            source (str or pystac.Link): The source from which to read.
+            root (Catalog or Collection): Optional root of the catalog for this object.
+                If provided, the root's resolved object cache can be used to search for
+                previously resolved instances of the STAC object.
+
+        Returns:
+            STACObject: The deserialized STACObject from the serialized JSON
+            contained in the file at the given uri.
+        """
+        d = self.read_json(source)
+        href = source if isinstance(source, str) else source.get_absolute_href()
+        return self.stac_object_from_dict(d, href=href, root=root)
+
+    def save_json(self, dest: Union[str, "Link_Type"], json_dict: Dict[str, Any]) -> None:
+        """Write a dict to the given URI as JSON.
+
+        See :func:`StacIO.write_text <pystac.StacIO.write_text>` for usage of
+        str vs Link as a parameter.
+
+        Args:
+            dest (str or pystac.Link): The destination file to write the text to.
+            json_dict (dict): The JSON dict to write.
+        """
+        txt = self._json_dumps(json_dict, dest)
+        self.write_text(dest, txt)
+
+    @classmethod
+    def set_default(cls, stac_io_class: Type["StacIO"]) -> None:
+        """Set the default StacIO instance to use."""
+        cls._default_io = stac_io_class
+
+    @classmethod
+    def default(cls) -> "StacIO":
+        if cls._default_io is None:
+            cls._default_io = DefaultStacIO
+
+        return cls._default_io()
+
+
+class DefaultStacIO(StacIO):
+    def read_text(self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any) -> 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}")
+
+        parsed = urlparse(href)
         if parsed.scheme != '':
             try:
-                with urlopen(uri) as f:
+                with urlopen(href) as f:
                     return f.read().decode('utf-8')
             except HTTPError as e:
-                raise Exception("Could not read uri {}".format(uri)) from e
+                raise Exception("Could not read uri {}".format(href)) from e
         else:
-            with open(uri) as f:
+            with open(href) as f:
                 return f.read()
 
-    @staticmethod
-    def default_write_text_method(uri: str, txt: str) -> None:
-        """Default method for writing text. Only handles local file paths."""
-        dirname = os.path.dirname(uri)
+    def write_text(self, dest: Union[str, "Link_Type"], txt: str, *args: Any,
+                   **kwargs: Any) -> None:
+        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}")
+
+        dirname = os.path.dirname(href)
         if dirname != '' and not os.path.isdir(dirname):
             os.makedirs(dirname)
-        with open(uri, 'w') as f:
+        with open(href, 'w') as f:
             f.write(txt)
 
-    read_text_method: Callable[[str], str] = default_read_text_method
-    """Users of PySTAC can replace the read_text_method in order
-    to expand the ability of PySTAC to read different file systems.
-    For example, a client of the library might replace this class
-    member in it's own __init__.py with a method that can read from
-    cloud storage.
+
+class DuplicateObjectKeyError(Exception):
+    pass
+
+
+class DuplicateKeyReportingMixin(StacIO):
+    """A mixin for StacIO implementations that will report
+    on duplicate keys in the JSON being read in.
+
+    See https://github.com/stac-utils/pystac/issues/313
     """
+    def _json_loads(self, txt: str, source: Union[str, "Link_Type"]) -> Dict[str, Any]:
+        return json.loads(txt, object_pairs_hook=self.duplicate_object_names_report_builder(source))
+
+    @staticmethod
+    def duplicate_object_names_report_builder(
+            source: Union[str, "Link_Type"]) -> Callable[[List[Tuple[str, Any]]], Dict[str, Any]]:
+        def report_duplicate_object_names(object_pairs: List[Tuple[str, Any]]) -> Dict[str, Any]:
+            result: Dict[str, Any] = {}
+            for key, value in object_pairs:
+                if key in result:
+                    url = source if isinstance(source, str) else source.get_absolute_href()
+                    raise DuplicateObjectKeyError(f"Found duplicate object name “{key}” in “{url}”")
+                else:
+                    result[key] = value
+            return result
+
+        return report_duplicate_object_names
+
+
+class STAC_IO:
+    """DEPRECATED: Methods used to read and save STAC json.
+    Allows users of the library to set their own methods
+    (e.g. for reading and writing from cloud storage)
 
-    write_text_method: Callable[[str, str], None] = default_write_text_method
-    """Users of PySTAC can replace the write_text_method in order
-    to expand the ability of PySTAC to write to different file systems.
-    For example, a client of the library might replace this class
-    member in it's own __init__.py with a method that can read from
-    cloud storage.
+    Note: The static methods of this class are deprecated. Move to using
+    instance methods of a specific instance of StacIO.
     """
+    @staticmethod
+    def read_text_method(uri: str) -> str:
+        return StacIO.default().read_text(uri)
+
+    @staticmethod
+    def write_text_method(uri: str, txt: str) -> None:
+        """Default method for writing text."""
+        return StacIO.default().write_text(uri, txt)
 
     @staticmethod
     def stac_object_from_dict(d: Dict[str, Any],