Implement #370 for EO Extension

Author: duckontheweb

CHANGELOG.md

@@ -12,6 +12,8 @@
   STAC Items ([#430](https://github.com/stac-utils/pystac/pull/430))
 - Support for Python 3.9 ([#420](https://github.com/stac-utils/pystac/pull/420))
 - Migration for pre-1.0.0-rc.1 Stats Objects (renamed to Range Objects in 1.0.0-rc.3) ([#447](https://github.com/stac-utils/pystac/pull/447))
+- Attempt to extend a `STACObject` that does not contain the extension's schema URI in
+  `stac_extensions` raises new `ExtensionNotImplementedError`.
 
 ### Changed
 

pystac/__init__.py

@@ -7,6 +7,7 @@
     STACError,
     STACTypeError,
     ExtensionAlreadyExistsError,
+    ExtensionNotImplemented,
     ExtensionTypeError,
     RequiredPropertyMissing,
     STACValidationError,

pystac/errors.py

@@ -35,6 +35,11 @@ class ExtensionAlreadyExistsError(Exception):
     pass
 
 
+class ExtensionNotImplemented(Exception):
+    """Attempted to extend a STAC object that does not implement the given
+    extension."""
+
+
 class RequiredPropertyMissing(Exception):
     """This error is raised when a required value was expected
     to be there but was missing or None. This will happen, for example,

pystac/extensions/base.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from typing import Generic, Iterable, List, Optional, Dict, Any, Type, TypeVar, Union
 
-from pystac import Collection, RangeSummary, STACObject, Summaries
+import pystac
 
 
 class SummariesExtension:
@@ -12,16 +12,16 @@ class SummariesExtension:
     extension-specific class that inherits from this class and instantiate that. See
     :class:`~pystac.extensions.eo.SummariesEOExtension` for an example."""
 
-    summaries: Summaries
+    summaries: pystac.Summaries
     """The summaries for the :class:`~pystac.Collection` being extended."""
 
-    def __init__(self, collection: Collection) -> None:
+    def __init__(self, collection: pystac.Collection) -> None:
         self.summaries = collection.summaries
 
     def _set_summary(
         self,
         prop_key: str,
-        v: Optional[Union[List[Any], RangeSummary[Any], Dict[str, Any]]],
+        v: Optional[Union[List[Any], pystac.RangeSummary[Any], Dict[str, Any]]],
     ) -> None:
         if v is None:
             self.summaries.remove(prop_key)
@@ -77,7 +77,7 @@ def _set_property(
             self.properties[prop_name] = v
 
 
-S = TypeVar("S", bound=STACObject)
+S = TypeVar("S", bound=pystac.STACObject)
 
 
 class ExtensionManagementMixin(Generic[S], ABC):
@@ -124,3 +124,17 @@ def has_extension(cls, obj: S) -> bool:
             obj.stac_extensions is not None
             and cls.get_schema_uri() in obj.stac_extensions
         )
+
+    @classmethod
+    def validate_has_extension(cls, obj: Union[S, pystac.Asset]) -> None:
+        """Given a :class:`~pystac.STACObject` or :class:`pystac.Asset` instance, checks
+        if the object (or its owner in the case of an Asset) has this extension's schema
+        URI in it's :attr:`~pystac.STACObject.stac_extensions` list."""
+        extensible = obj.owner if isinstance(obj, pystac.Asset) else obj
+        if (
+            extensible is not None
+            and cls.get_schema_uri() not in extensible.stac_extensions
+        ):
+            raise pystac.ExtensionNotImplemented(
+                f"Could not find extension schema URI {cls.get_schema_uri()} in object."
+            )

pystac/extensions/eo.py

@@ -346,8 +346,8 @@ def cloud_cover(self, v: Optional[float]) -> None:
     def get_schema_uri(cls) -> str:
         return SCHEMA_URI
 
-    @staticmethod
-    def ext(obj: T) -> "EOExtension[T]":
+    @classmethod
+    def ext(cls, obj: T) -> "EOExtension[T]":
         """Extends the given STAC Object with properties from the :stac-ext:`Electro-Optical
         Extension <eo>`.
 
@@ -359,8 +359,10 @@ def ext(obj: T) -> "EOExtension[T]":
             pystac.ExtensionTypeError : If an invalid object type is passed.
         """
         if isinstance(obj, pystac.Item):
+            cls.validate_has_extension(obj)
             return cast(EOExtension[T], ItemEOExtension(obj))
         elif isinstance(obj, pystac.Asset):
+            cls.validate_has_extension(obj)
             return cast(EOExtension[T], AssetEOExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

tests/extensions/test_eo.py

@@ -264,3 +264,23 @@ def test_extend_invalid_object(self) -> None:
 
         with self.assertRaises(pystac.ExtensionTypeError):
             EOExtension.ext(link)  # type: ignore
+
+    def test_extension_not_implemented(self) -> None:
+        # Should raise exception if Item does not include extension URI
+        item = pystac.Item.from_file(self.PLAIN_ITEM)
+
+        with self.assertRaises(pystac.ExtensionNotImplemented):
+            _ = EOExtension.ext(item)
+
+        # Should raise exception if owning Item does not include extension URI
+        asset = item.assets["thumbnail"]
+
+        with self.assertRaises(pystac.ExtensionNotImplemented):
+            _ = EOExtension.ext(asset)
+
+        # Should succeed if Asset has no owner
+        stac_io = pystac.StacIO.default()
+        item_dict = stac_io.read_json(self.LANDSAT_EXAMPLE_URI)
+        asset_dict = item_dict["assets"]["B11"]
+        ownerless_asset = pystac.Asset.from_dict(asset_dict)
+        _ = EOExtension.ext(ownerless_asset)