Simplify from_file logic

Author: duckontheweb

pystac/__init__.py

@@ -76,10 +76,11 @@
 def read_file(href: str) -> STACObject:
     """Reads a STAC object from a file.
 
-    This method will return either a Catalog, a Collection, or an Item based on what the
-    file contains.
+    This method will return either a Catalog, a Collection, or an Item based on what
+    the file contains.
 
-    This is a convenience method for :meth:`STACObject.from_file <pystac.STACObject.from_file>`
+    This is a convenience method for :meth:`StacIO.read_stac_object
+    <pystac.StacIO.read_stac_object>`
 
     Args:
         href : The HREF to read the object from.
@@ -90,22 +91,12 @@ def read_file(href: str) -> STACObject:
 
     Raises:
         STACTypeError : If the file at ``href`` does not represent a valid
-            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection` is not
-            a :class:`~pystac.STACObject` and must be read using
+            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection`
+            is not a :class:`~pystac.STACObject` and must be read using
             :meth:`ItemCollection.from_file <pystac.ItemCollection.from_file>`
     """
     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)
-    elif typ == STACObjectType.COLLECTION:
-        return Collection.from_file(href)
-    elif typ == STACObjectType.ITEM:
-        return Item.from_file(href)
-    else:
-        raise STACTypeError(f"Cannot read file of type {typ}")
+    return stac_io.read_stac_object(href)
 
 
 def write_file(
@@ -148,7 +139,7 @@ def read_dict(
     or :class`~Item` based on the contents of the dict.
 
     This is a convenience method for either
-    :meth:`stac_io.stac_object_from_dict <stac_io.stac_object_from_dict>`.
+    :meth:`StacIO.stac_object_from_dict <pystac.StacIO.stac_object_from_dict>`.
 
     Args:
         d : The dict to parse.
@@ -162,8 +153,8 @@ def read_dict(
 
     Raises:
         STACTypeError : If the ``d`` dictionary does not represent a valid
-            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection` is not
-            a :class:`~pystac.STACObject` and must be read using
+            :class:`~pystac.STACObject`. Note that an :class:`~pystac.ItemCollection`
+            is not a :class:`~pystac.STACObject` and must be read using
             :meth:`ItemCollection.from_dict <pystac.ItemCollection.from_dict>`
     """
     if stac_io is None:

pystac/catalog.py

@@ -1,6 +1,7 @@
 import os
 from copy import deepcopy
 from enum import Enum
+from pystac.errors import STACTypeError
 from typing import (
     Any,
     Callable,
@@ -15,14 +16,19 @@
 )
 
 import pystac
-from pystac.stac_object import STACObject
+from pystac.stac_object import STACObject, STACObjectType
 from pystac.layout import (
     BestPracticesLayoutStrategy,
     HrefLayoutStrategy,
     LayoutTemplate,
 )
 from pystac.link import Link
 from pystac.cache import ResolvedObjectCache
+from pystac.serialization import (
+    identify_stac_object_type,
+    identify_stac_object,
+    migrate_to_latest,
+)
 from pystac.utils import is_absolute_href, make_absolute_href
 
 if TYPE_CHECKING:
@@ -902,10 +908,11 @@ def from_dict(
         migrate: bool = False,
     ) -> "Catalog":
         if migrate:
-            result = pystac.read_dict(d, href=href, root=root)
-            if not isinstance(result, Catalog):
-                raise pystac.STACTypeError(f"{result} is not a Catalog")
-            return result
+            info = identify_stac_object(d)
+            d = migrate_to_latest(d, info)
+
+        if not cls.identify_dict(d):
+            raise STACTypeError(f"{d} does not represent a {cls.__name__} instance")
 
         catalog_type = CatalogType.determine_type(d)
 
@@ -955,3 +962,7 @@ def from_file(cls, href: str, stac_io: Optional[pystac.StacIO] = None) -> "Catal
         result._stac_io = stac_io
 
         return result
+
+    @classmethod
+    def identify_dict(cls, d: Dict[str, Any]) -> bool:
+        return identify_stac_object_type(d) == STACObjectType.CATALOG

pystac/collection.py

@@ -1,5 +1,6 @@
 from copy import copy, deepcopy
 from datetime import datetime as Datetime
+from pystac.errors import STACTypeError
 from typing import (
     Any,
     Dict,
@@ -23,6 +24,11 @@
 from pystac.layout import HrefLayoutStrategy
 from pystac.link import Link
 from pystac.utils import datetime_to_str
+from pystac.serialization import (
+    identify_stac_object_type,
+    identify_stac_object,
+    migrate_to_latest,
+)
 from pystac.summaries import Summaries
 
 if TYPE_CHECKING:
@@ -583,10 +589,11 @@ def from_dict(
         migrate: bool = False,
     ) -> "Collection":
         if migrate:
-            result = pystac.read_dict(d, href=href, root=root)
-            if not isinstance(result, Collection):
-                raise pystac.STACError(f"{result} is not a Catalog")
-            return result
+            info = identify_stac_object(d)
+            d = migrate_to_latest(d, info)
+
+        if not cls.identify_dict(d):
+            raise STACTypeError(f"{d} does not represent a {cls.__name__} instance")
 
         catalog_type = CatalogType.determine_type(d)
 
@@ -676,3 +683,7 @@ def from_file(
         if not isinstance(result, Collection):
             raise pystac.STACTypeError(f"{result} is not a {Collection}.")
         return result
+
+    @classmethod
+    def identify_dict(cls, d: Dict[str, Any]) -> bool:
+        return identify_stac_object_type(d) == STACObjectType.COLLECTION

pystac/item.py

@@ -9,6 +9,11 @@
 from pystac import STACError, STACObjectType
 from pystac.asset import Asset
 from pystac.link import Link
+from pystac.serialization import (
+    identify_stac_object_type,
+    identify_stac_object,
+    migrate_to_latest,
+)
 from pystac.stac_object import STACObject
 from pystac.utils import (
     is_absolute_href,
@@ -912,10 +917,13 @@ 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)
+
+        if not cls.identify_dict(d):
+            raise pystac.STACTypeError(
+                f"{d} does not represent a {cls.__name__} instance"
+            )
 
         d = deepcopy(d)
         id = d.pop("id")
@@ -980,3 +988,7 @@ def from_file(cls, href: str, stac_io: Optional[pystac.StacIO] = None) -> "Item"
         if not isinstance(result, Item):
             raise pystac.STACTypeError(f"{result} is not a {Item}.")
         return result
+
+    @classmethod
+    def identify_dict(cls, d: Dict[str, Any]) -> bool:
+        return identify_stac_object_type(d) == STACObjectType.ITEM

pystac/stac_object.py

@@ -6,8 +6,6 @@
 from pystac import STACError
 from pystac.link import Link
 from pystac.utils import is_absolute_href, make_absolute_href
-from pystac import serialization
-from pystac.serialization.identify import identify_stac_object
 
 if TYPE_CHECKING:
     from pystac.catalog import Catalog as Catalog_Type
@@ -465,16 +463,17 @@ def from_file(
             The specific STACObject implementation class that is represented
             by the JSON read from the file located at HREF.
         """
+        if cls == STACObject:
+            return pystac.read_file(href)
+
         if stac_io is None:
             stac_io = pystac.StacIO.default()
 
         if not is_absolute_href(href):
             href = make_absolute_href(href)
 
         d = stac_io.read_json(href)
-        info = identify_stac_object(d)
-        d = serialization.migrate.migrate_to_latest(d, info)
-        o = cls.from_dict(d, href=href)
+        o = cls.from_dict(d, href=href, migrate=True)
 
         # Set the self HREF, if it's not already set to something else.
         if o.get_self_href() is None:
@@ -513,3 +512,16 @@ def from_dict(
             STACObject: The STACObject parsed from this dict.
         """
         pass
+
+    @classmethod
+    @abstractmethod
+    def identify_dict(cls, d: Dict[str, Any]) -> bool:
+        """Returns a boolean indicating whether the given dictionary represents a valid
+        instance of this :class:`~STACObject` sub-class.
+
+        Args:
+            d : A dictionary to identify
+        """
+        raise NotImplementedError(
+            "identify_dict must be implemented by the STACObject subclass."
+        )

tests/test_catalog.py

@@ -974,6 +974,14 @@ def test_catalog_with_href_caches_by_href(self) -> None:
         # cached only by HREF
         self.assertEqual(len(cache.id_keys_to_objects), 0)
 
+    def testfrom_invalid_dict_raises_exception(self) -> None:
+        stac_io = pystac.StacIO.default()
+        collection_dict = stac_io.read_json(
+            TestCases.get_path("data-files/collections/multi-extent.json")
+        )
+        with self.assertRaises(pystac.STACTypeError):
+            _ = pystac.Catalog.from_dict(collection_dict)
+
 
 class FullCopyTest(unittest.TestCase):
     def check_link(self, link: pystac.Link, tag: str) -> None:

tests/test_collection.py

@@ -196,6 +196,14 @@ def test_schema_summary(self) -> None:
 
         self.assertIsInstance(instruments_schema, dict)
 
+    def test_from_invalid_dict_raises_exception(self) -> None:
+        stac_io = pystac.StacIO.default()
+        catalog_dict = stac_io.read_json(
+            TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
+        )
+        with self.assertRaises(pystac.STACTypeError):
+            _ = pystac.Collection.from_dict(catalog_dict)
+
 
 class ExtentTest(unittest.TestCase):
     def test_spatial_allows_single_bbox(self) -> None:

tests/test_item.py

@@ -216,6 +216,14 @@ def test_make_asset_href_relative_is_noop_on_relative_hrefs(self) -> None:
         item.make_asset_hrefs_relative()
         self.assertEqual(asset.get_absolute_href(), original_href)
 
+    def test_from_invalid_dict_raises_exception(self) -> None:
+        stac_io = pystac.StacIO.default()
+        catalog_dict = stac_io.read_json(
+            TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
+        )
+        with self.assertRaises(pystac.STACTypeError):
+            _ = pystac.Item.from_dict(catalog_dict)
+
 
 class CommonMetadataTest(unittest.TestCase):
     def setUp(self) -> None:

tests/test_writing.py

@@ -72,7 +72,9 @@ def validate_catalog_link_type(
             href: str, link_type: str, should_include_self: bool
         ) -> None:
             cat_dict = pystac.StacIO.default().read_json(href)
-            cat = pystac.Catalog.from_file(href)
+            cat = pystac.read_file(href)
+            if not isinstance(cat, pystac.Catalog):
+                raise pystac.STACTypeError(f"File at {href} is not a Catalog.")
 
             rels = set([link["rel"] for link in cat_dict["links"]])
             self.assertEqual("self" in rels, should_include_self)