[WIP] Improve STACObject inheritance

docs/api.rst

@@ -164,11 +164,18 @@ RelType
 IO
 --
 
+StacIO
+~~~~~~
+
+.. autoclass:: pystac.StacIO
+   :members:
+   :undoc-members:
+
 STAC_IO
 ~~~~~~~
 
-STAC_IO is the utility mechanism that PySTAC uses for reading and writing. Users of
-PySTAC can hook into PySTAC by overriding members to utilize their own IO methods.
+.. deprecated:: 1.0.0-beta.1
+   Use :class:`pystac.StacIO` instead. This class will be removed in v1.0.0.
 
 .. autoclass:: pystac.stac_io.STAC_IO
    :members:

pystac/__init__.py

@@ -72,7 +72,7 @@
 )
 
 
-def read_file(href: str) -> STACObject:
+def read_file(href: str, migrate: bool = False) -> STACObject:
     """Reads a STAC object from a file.
 
     This method will return either a Catalog, a Collection, or an Item based on what the
@@ -82,6 +82,8 @@ def read_file(href: str) -> STACObject:
 
     Args:
         href : The HREF to read the object from.
+        migrate: Indicates whether to migrate the object to the latest
+                version when reading from file. Defaults to ``False``.
 
     Returns:
         The specific STACObject implementation class that is represented
@@ -93,7 +95,18 @@ def read_file(href: str) -> STACObject:
             a :class:`~pystac.STACObject` and must be read using
             :meth:`ItemCollection.from_file <pystac.ItemCollection.from_file>`
     """
-    return STACObject.from_file(href)
+    stac_io = StacIO.default()
+    d = stac_io.read_json(href)
+    typ = pystac.serialization.identify.identify_stac_object_type(d)
+
+    if typ == STACObjectType.CATALOG:
+        return Catalog.from_file(href, migrate=migrate)
+    elif typ == STACObjectType.COLLECTION:
+        return Collection.from_file(href, migrate=migrate)
+    elif typ == STACObjectType.ITEM:
+        return Item.from_file(href, migrate=migrate)
+    else:
+        raise STACTypeError(f"Cannot read file of type {typ}")
 
 
 def write_file(
@@ -128,6 +141,7 @@ def read_dict(
     href: Optional[str] = None,
     root: Optional[Catalog] = None,
     stac_io: Optional[StacIO] = None,
+    migrate: bool = False,
 ) -> STACObject:
     """Reads a :class:`~STACObject` or :class:`~ItemCollection` from a JSON-like dict
     representing a serialized STAC object.
@@ -156,4 +170,4 @@ def read_dict(
     """
     if stac_io is None:
         stac_io = StacIO.default()
-    return stac_io.stac_object_from_dict(d, href, root)
+    return stac_io.stac_object_from_dict(d, href, root, migrate)

pystac/catalog.py

@@ -904,7 +904,7 @@ def from_dict(
         if migrate:
             result = pystac.read_dict(d, href=href, root=root)
             if not isinstance(result, Catalog):
-                raise pystac.STACError(f"{result} is not a Catalog")
+                raise pystac.STACTypeError(f"{result} is not a Catalog")
             return result
 
         catalog_type = CatalogType.determine_type(d)
@@ -919,7 +919,7 @@ def from_dict(
 
         d.pop("stac_version")
 
-        cat = Catalog(
+        cat = cls(
             id=id,
             description=description,
             title=title,
@@ -945,8 +945,18 @@ def full_copy(
         return cast(Catalog, super().full_copy(root, parent))
 
     @classmethod
-    def from_file(cls, href: str, stac_io: Optional[pystac.StacIO] = None) -> "Catalog":
-        result = super().from_file(href, stac_io)
+    def from_file(
+        cls,
+        href: str,
+        stac_io: Optional[pystac.StacIO] = None,
+        migrate: bool = False,
+    ) -> "Catalog":
+        if stac_io is None:
+            stac_io = pystac.StacIO.default()
+
+        result = super().from_file(href, stac_io, migrate)
         if not isinstance(result, Catalog):
             raise pystac.STACTypeError(f"{result} is not a {Catalog}.")
+        result._stac_io = stac_io
+
         return result

pystac/collection.py

@@ -610,7 +610,7 @@ def from_dict(
 
         d.pop("stac_version")
 
-        collection = Collection(
+        collection = cls(
             id=id,
             description=description,
             extent=extent,
@@ -670,9 +670,12 @@ def full_copy(
 
     @classmethod
     def from_file(
-        cls, href: str, stac_io: Optional[pystac.StacIO] = None
+        cls,
+        href: str,
+        stac_io: Optional[pystac.StacIO] = None,
+        migrate: bool = False,
     ) -> "Collection":
-        result = super().from_file(href, stac_io)
+        result = super().from_file(href, stac_io, migrate)
         if not isinstance(result, Collection):
             raise pystac.STACTypeError(f"{result} is not a {Collection}.")
         return result

pystac/item.py

@@ -9,6 +9,8 @@
 from pystac import STACError, STACObjectType
 from pystac.asset import Asset
 from pystac.link import Link
+from pystac.serialization.migrate import migrate_to_latest
+from pystac.serialization.identify import identify_stac_object
 from pystac.stac_object import STACObject
 from pystac.utils import (
     is_absolute_href,
@@ -912,10 +914,8 @@ def from_dict(
         migrate: bool = False,
     ) -> "Item":
         if migrate:
-            result = pystac.read_dict(d, href=href, root=root)
-            if not isinstance(result, Item):
-                raise pystac.STACError(f"{result} is not a Catalog")
-            return result
+            info = identify_stac_object(d)
+            d = migrate_to_latest(d, info)
 
         d = deepcopy(d)
         id = d.pop("id")
@@ -975,8 +975,13 @@ def full_copy(
         return cast(Item, super().full_copy(root, parent))
 
     @classmethod
-    def from_file(cls, href: str, stac_io: Optional[pystac.StacIO] = None) -> "Item":
-        result = super().from_file(href, stac_io)
+    def from_file(
+        cls,
+        href: str,
+        stac_io: Optional[pystac.StacIO] = None,
+        migrate: bool = False,
+    ) -> "Item":
+        result = super().from_file(href, stac_io, migrate)
         if not isinstance(result, Item):
             raise pystac.STACTypeError(f"{result} is not a {Item}.")
         return result

pystac/serialization/__init__.py

@@ -1,54 +1,8 @@
 # flake8: noqa
-from typing import Any, Dict, Optional, TYPE_CHECKING
-
-import pystac
 from pystac.serialization.identify import (
     STACVersionRange,
     identify_stac_object,
     identify_stac_object_type,
 )
 from pystac.serialization.common_properties import merge_common_properties
 from pystac.serialization.migrate import migrate_to_latest
-
-if TYPE_CHECKING:
-    from pystac.stac_object import STACObject
-    from pystac.catalog import Catalog
-
-
-def stac_object_from_dict(
-    d: Dict[str, Any], href: Optional[str] = None, root: Optional["Catalog"] = None
-) -> "STACObject":
-    """Determines how to deserialize a dictionary into a STAC object.
-
-    Args:
-        d : The dict to parse.
-        href : Optional href that is the file location of the object being
-            parsed.
-        root : 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.
-
-    Note: This is used internally in StacIO instances to deserialize STAC Objects.
-    """
-    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)
-
-    info = identify_stac_object(d)
-
-    d = migrate_to_latest(d, info)
-
-    if info.object_type == pystac.STACObjectType.CATALOG:
-        return pystac.Catalog.from_dict(d, href=href, root=root, migrate=False)
-
-    if info.object_type == pystac.STACObjectType.COLLECTION:
-        return pystac.Collection.from_dict(d, href=href, root=root, migrate=False)
-
-    if info.object_type == pystac.STACObjectType.ITEM:
-        return pystac.Item.from_dict(d, href=href, root=root, migrate=False)
-
-    raise pystac.STACTypeError(f"Unknown STAC object type {info.object_type}")

pystac/stac_io.py

@@ -19,7 +19,12 @@
 
 import pystac
 from pystac.utils import safe_urlparse
-import pystac.serialization
+from pystac.serialization import (
+    merge_common_properties,
+    identify_stac_object_type,
+    identify_stac_object,
+    migrate_to_latest,
+)
 
 # Use orjson if available
 try:
@@ -94,13 +99,35 @@ def stac_object_from_dict(
         d: Dict[str, Any],
         href: Optional[str] = None,
         root: Optional["Catalog_Type"] = None,
+        migrate: bool = False,
     ) -> "STACObject_Type":
-        result = pystac.serialization.stac_object_from_dict(d, href, root)
-        if isinstance(result, pystac.Catalog):
-            # Set the stac_io instance for usage by io operations
-            # where this catalog is the root.
+        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
+            )
+
+        info = identify_stac_object(d)
+
+        if migrate:
+            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)
             result._stac_io = self
-        return result
+            return result
+
+        if info.object_type == pystac.STACObjectType.COLLECTION:
+            return pystac.Collection.from_dict(d, href=href, root=root, migrate=False)
+
+        if info.object_type == pystac.STACObjectType.ITEM:
+            return pystac.Item.from_dict(d, href=href, root=root, migrate=False)
+
+        raise ValueError(f"Unknown STAC object type {info.object_type}")
 
     def read_json(
         self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any
@@ -121,7 +148,10 @@ def read_json(
         return self._json_loads(txt, source)
 
     def read_stac_object(
-        self, source: Union[str, "Link_Type"], root: Optional["Catalog_Type"] = None
+        self,
+        source: Union[str, "Link_Type"],
+        root: Optional["Catalog_Type"] = None,
+        migrate: bool = False,
     ) -> "STACObject_Type":
         """Read a STACObject from a JSON file at the given source.
 
@@ -140,7 +170,7 @@ def read_stac_object(
         """
         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)
+        return self.stac_object_from_dict(d, href=href, root=root, migrate=migrate)
 
     def save_json(
         self, dest: Union[str, "Link_Type"], json_dict: Dict[str, Any]
@@ -271,36 +301,63 @@ class STAC_IO:
     """
 
     @staticmethod
-    def read_text_method(uri: str) -> str:
+    def issue_deprecation_warning() -> None:
         warnings.warn(
-            "STAC_IO is deprecated. "
-            "Please use instances of StacIO (e.g. StacIO.default()).",
+            "STAC_IO is deprecated and will be removed in v1.0.0. "
+            "Please use instances of StacIO (e.g. StacIO.default()) instead.",
             DeprecationWarning,
         )
+
+    def __init__(self) -> None:
+        STAC_IO.issue_deprecation_warning()
+
+    def __init_subclass__(cls) -> None:
+        STAC_IO.issue_deprecation_warning()
+
+    @staticmethod
+    def read_text_method(uri: str) -> str:
+        STAC_IO.issue_deprecation_warning()
         return StacIO.default().read_text(uri)
 
     @staticmethod
     def write_text_method(uri: str, txt: str) -> None:
         """Default method for writing text."""
-        warnings.warn(
-            "STAC_IO is deprecated. "
-            "Please use instances of StacIO (e.g. StacIO.default()).",
-            DeprecationWarning,
-        )
+        STAC_IO.issue_deprecation_warning()
         return StacIO.default().write_text(uri, txt)
 
     @staticmethod
     def stac_object_from_dict(
         d: Dict[str, Any],
         href: Optional[str] = None,
         root: Optional["Catalog_Type"] = None,
+        migrate: bool = False,
     ) -> "STACObject_Type":
-        warnings.warn(
-            "STAC_IO is deprecated. "
-            "Please use instances of StacIO (e.g. StacIO.default()).",
-            DeprecationWarning,
-        )
-        return pystac.serialization.stac_object_from_dict(d, href, root)
+        STAC_IO.issue_deprecation_warning()
+        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
+            )
+
+        info = identify_stac_object(d)
+
+        if migrate:
+            d = migrate_to_latest(d, info)
+
+        if info.object_type == pystac.STACObjectType.CATALOG:
+            return pystac.Catalog.from_dict(d, href=href, root=root, migrate=False)
+
+        if info.object_type == pystac.STACObjectType.COLLECTION:
+            return pystac.Collection.from_dict(d, href=href, root=root, migrate=False)
+
+        if info.object_type == pystac.STACObjectType.ITEM:
+            return pystac.Item.from_dict(d, href=href, root=root, migrate=False)
+
+        raise ValueError(f"Unknown STAC object type {info.object_type}")
 
     # This is set in __init__.py
     _STAC_OBJECT_CLASSES = None
@@ -356,6 +413,7 @@ def read_json(cls, uri: str) -> Dict[str, Any]:
             STAC_IO in order to enable additional URI types, replace that member
             with your own implementation.
         """
+        STAC_IO.issue_deprecation_warning()
         result: Dict[str, Any] = json.loads(STAC_IO.read_text(uri))
         return result