Separate method to validate Asset owner extensions

Author: duckontheweb

pystac/extensions/base.py

@@ -1,5 +1,16 @@
 from abc import ABC, abstractmethod
-from typing import Generic, Iterable, List, Optional, Dict, Any, Type, TypeVar, Union
+from typing import (
+    cast,
+    Generic,
+    Iterable,
+    List,
+    Optional,
+    Dict,
+    Any,
+    Type,
+    TypeVar,
+    Union,
+)
 
 import pystac
 
@@ -126,29 +137,40 @@ def has_extension(cls, obj: S) -> bool:
         )
 
     @classmethod
-    def validate_has_extension(cls, obj: Optional[S], add_if_missing: bool) -> None:
+    def validate_owner_has_extension(
+        cls, asset: pystac.Asset, add_if_missing: bool
+    ) -> None:
+        """Given an :class:`~pystac.Asset`, checks if the asset's owner has this
+        extension's schema URI in its :attr:`~pystac.STACObject.stac_extensions` list.
+        If ``add_if_missing`` is ``True``, the schema URI will be added to the owner.
+
+        Raises:
+            STACError : If ``add_if_missing`` is ``True`` and ``asset.owner`` is
+                ``None``.
+        """
+        if asset.owner is None:
+            if add_if_missing:
+                raise pystac.STACError(
+                    "Can only add schema URIs to Assets with an owner."
+                )
+            else:
+                return
+        return cls.validate_has_extension(cast(S, asset.owner), add_if_missing)
+
+    @classmethod
+    def validate_has_extension(cls, obj: S, add_if_missing: bool) -> None:
         """Given a :class:`~pystac.STACObject`, checks if the object has this
-        extension's schema URI in it's :attr:`~pystac.STACObject.stac_extensions` list.
+        extension's schema URI in its :attr:`~pystac.STACObject.stac_extensions` list.
         If ``add_if_missing`` is ``True``, the schema URI will be added to the object.
 
         Args:
             obj : The object to validate.
-            add_if_missing : Whether to add the schema URI to the object if the URI is
+            add_if_missing : Whether to add the schema URI to the object if it is
                 not already present.
-
-        Raises:
-            STACError : If ``add_if_missing`` is ``True`` and ``extensible`` is None.
         """
         if add_if_missing:
-            if obj is None:
-                raise pystac.STACError(
-                    "Can only add schema URIs to Assets with an owner."
-                )
             cls.add_to(obj)
 
-        if obj is None:
-            return
-
         if cls.get_schema_uri() not in obj.stac_extensions:
             raise pystac.ExtensionNotImplemented(
                 f"Could not find extension schema URI {cls.get_schema_uri()} in object."

pystac/extensions/datacube.py

@@ -346,7 +346,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "DatacubeExtension[T]":
             cls.validate_has_extension(obj, add_if_missing)
             return cast(DatacubeExtension[T], ItemDatacubeExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(DatacubeExtension[T], AssetDatacubeExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/eo.py

@@ -364,7 +364,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "EOExtension[T]":
             cls.validate_has_extension(obj, add_if_missing)
             return cast(EOExtension[T], ItemEOExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(EOExtension[T], AssetEOExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/file.py

@@ -199,7 +199,7 @@ def ext(cls, obj: pystac.Asset, add_if_missing: bool = False) -> "FileExtension"
         This extension can be applied to instances of :class:`~pystac.Asset`.
         """
         if isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cls(obj)
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/pointcloud.py

@@ -530,10 +530,9 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "PointcloudExtension[T]":
         elif isinstance(obj, pystac.Asset):
             if obj.owner is not None and not isinstance(obj.owner, pystac.Item):
                 raise pystac.ExtensionTypeError(
-                    "Pointcloud extension does not apply to Assets owned by anything "
-                    "other than an Item."
+                    "Pointcloud extension does not apply to Collection Assets."
                 )
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(PointcloudExtension[T], AssetPointcloudExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/projection.py

@@ -275,7 +275,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "ProjectionExtension[T]":
             cls.validate_has_extension(obj, add_if_missing)
             return cast(ProjectionExtension[T], ItemProjectionExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(ProjectionExtension[T], AssetProjectionExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/raster.py

@@ -704,7 +704,7 @@ def ext(cls, obj: pystac.Asset, add_if_missing: bool = False) -> "RasterExtensio
             pystac.ExtensionTypeError : If an invalid object type is passed.
         """
         if isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cls(obj)
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/sar.py

@@ -320,10 +320,9 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "SarExtension[T]":
         elif isinstance(obj, pystac.Asset):
             if obj.owner is not None and not isinstance(obj.owner, pystac.Item):
                 raise pystac.ExtensionTypeError(
-                    "SAR extension does not apply to Assets owned by anything "
-                    "other than an Item."
+                    "SAR extension does not apply to Collection Assets."
                 )
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(SarExtension[T], AssetSarExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/sat.py

@@ -152,7 +152,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "SatExtension[T]":
             cls.validate_has_extension(obj, add_if_missing)
             return cast(SatExtension[T], ItemSatExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(SatExtension[T], AssetSatExtension(obj))
         else:
             raise pystac.ExtensionTypeError(

pystac/extensions/timestamps.py

@@ -132,7 +132,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> "TimestampsExtension[T]":
             cls.validate_has_extension(obj, add_if_missing)
             return cast(TimestampsExtension[T], ItemTimestampsExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_has_extension(obj.owner, add_if_missing)
+            cls.validate_owner_has_extension(obj, add_if_missing)
             return cast(TimestampsExtension[T], AssetTimestampsExtension(obj))
         else:
             raise pystac.ExtensionTypeError(