Fix ID for invalid objects with stac_version.

Also adds "type" field in Collection/Catalog migration

Author: duckontheweb

pystac/serialization/identify.py

@@ -31,6 +31,15 @@ class OldExtensionShortIDs(Enum):
     FILE = "file"
 
 
+class STACType(str, Enum):
+    def __str__(self) -> str:
+        return str(self.value)
+
+    CATALOG = "Catalog"
+    COLLECTION = "Collection"
+    ITEM = "Feature"
+
+
 @total_ordering
 class STACVersionID:
     """Defines STAC versions in an object that is orderable based on version number.
@@ -180,17 +189,34 @@ def identify_stac_object_type(
     Args:
         json_dict : The dict of JSON to identify.
     """
-    # Try to identify using 'type' property, if present
-    if "type" in json_dict:
-        # Try to find 'type' property in known STACObjectType values
-        for t in pystac.STACObjectType:
-            if json_dict["type"].lower() == t.value.lower():
-                return t
-
+    stac_version = (
+        STACVersionID(json_dict["stac_version"])
+        if "stac_version" in json_dict
+        else None
+    )
     obj_type = json_dict.get("type")
 
+    # Try to identify using 'type' property for v1.0.0-rc.1 and higher
+    introduced_type_attribute = STACVersionID("1.0.0-rc.1")
+    if stac_version is not None and stac_version >= introduced_type_attribute:
+
+        # Since v1.0.0-rc.1 requires a "type" field for all STAC objects, any object
+        # that is missing this attribute is not a valid STAC object.
+        if obj_type is None:
+            return None
+
+        # Try to match the "type" attribute
+        if obj_type == STACType.CATALOG:
+            return pystac.STACObjectType.CATALOG
+        elif obj_type == STACType.COLLECTION:
+            return pystac.STACObjectType.COLLECTION
+        elif obj_type == STACType.ITEM:
+            return pystac.STACObjectType.ITEM
+        else:
+            return None
+
     # For pre-1.0 objects for version 0.8.* or later 'stac_version' must be present
-    if "stac_version" in json_dict:
+    if stac_version is not None:
         # Pre-1.0 STAC objects with 'type' == "Feature" are Items
         if obj_type == "Feature":
             return pystac.STACObjectType.ITEM

pystac/serialization/migrate.py

@@ -7,6 +7,7 @@
     OldExtensionShortIDs,
     STACJSONDescription,
     STACVersionID,
+    STACType,
 )
 
 if TYPE_CHECKING:
@@ -18,6 +19,7 @@ def _migrate_catalog(
 ) -> None:
     if version < "0.8":
         d["stac_extensions"] = list(info.extensions)
+    d["type"] = STACType.CATALOG
 
 
 def _migrate_collection_summaries(
@@ -36,6 +38,7 @@ def _migrate_collection(
     d: Dict[str, Any], version: STACVersionID, info: STACJSONDescription
 ) -> None:
     _migrate_catalog(d, version, info)
+    d["type"] = STACType.COLLECTION
     _migrate_collection_summaries(d, version, info)
 
 

tests/serialization/test_identify.py

@@ -56,6 +56,31 @@ def test_identify_non_stac_type(self) -> None:
 
         self.assertIsNone(identify_stac_object_type(plain_feature_dict))
 
+    def test_identify_invalid_stac_object_with_version(self) -> None:
+        # Has stac_version but is not a valid STAC object
+        invalid_dict = {
+            "id": "concepts",
+            "title": "Concepts catalogs",
+            "links": [
+                {
+                    "rel": "self",
+                    "type": "application/json",
+                    "href": "https://tamn.snapplanet.io/catalogs/concepts",
+                },
+                {
+                    "rel": "root",
+                    "type": "application/json",
+                    "href": "https://tamn.snapplanet.io",
+                },
+            ],
+            "stac_version": "1.0.0",
+        }
+
+        with self.assertRaises(pystac.STACTypeError) as ctx:
+            identify_stac_object(invalid_dict)
+
+        self.assertIn("JSON does not represent a STAC object", str(ctx.exception))
+
     def test_identify_non_stac_raises_error(self) -> None:
         plain_feature_dict = {
             "type": "Feature",