Assets keep their own stac_extensions

pystac/asset.py

@@ -35,6 +35,8 @@ class Asset:
         extra_fields : Optional, additional fields for this asset. This is used
             by extensions as a way to serialize and deserialize properties on asset
             object JSON.
+        stac_extensions : Optional, a list of schema URIs for STAC Extensions
+            implemented by this STAC Asset.
     """
 
     href: str
@@ -64,6 +66,9 @@ class Asset:
     """Optional, additional fields for this asset. This is used by extensions as a
     way to serialize and deserialize properties on asset object JSON."""
 
+    _stac_extensions: List[str]
+    """A list of schema URIs for STAC Extensions implemented by this STAC Asset."""
+
     def __init__(
         self,
         href: str,
@@ -72,13 +77,15 @@ def __init__(
         media_type: Optional[str] = None,
         roles: Optional[List[str]] = None,
         extra_fields: Optional[Dict[str, Any]] = None,
+        stac_extensions: Optional[List[str]] = None,
     ) -> None:
         self.href = utils.make_posix_style(href)
         self.title = title
         self.description = description
         self.media_type = media_type
         self.roles = roles
         self.extra_fields = extra_fields or {}
+        self._stac_extensions = stac_extensions
 
         # The Item which owns this Asset.
         self.owner = None

pystac/collection.py

@@ -577,8 +577,22 @@ def to_dict(
         )
         d["extent"] = self.extent.to_dict()
         d["license"] = self.license
+
+        # we use the fact that in recent Python versions, dict keys are ordered
+        # by default
+        stac_extensions: Optional[Dict[str, None]] = None
         if self.stac_extensions:
-            d["stac_extensions"] = self.stac_extensions
+            stac_extensions = dict.fromkeys(self.stac_extensions)
+
+        for asset in self.assets.values():
+            if stac_extensions and asset._stac_extensions:
+                stac_extensions.update(dict.fromkeys(asset._stac_extensions))
+            elif asset._stac_extensions:
+                stac_extensions = dict.fromkeys(asset._stac_extensions)
+
+        if stac_extensions is not None:
+            d["stac_extensions"] = list(stac_extensions.keys())
+
         if self.keywords:
             d["keywords"] = self.keywords
         if self.providers:

pystac/extensions/base.py

@@ -11,7 +11,7 @@
     Type,
     TypeVar,
     Union,
-    cast,
+    Protocol,
 )
 
 import pystac
@@ -96,7 +96,7 @@ def _set_property(
             self.properties[prop_name] = v
 
 
-S = TypeVar("S", bound=pystac.STACObject)
+S = TypeVar("S", bound=Union[pystac.STACObject, pystac.Asset])
 
 
 class ExtensionManagementMixin(Generic[S], ABC):
@@ -130,26 +130,58 @@ def add_to(cls, obj: S) -> None:
         """Add the schema URI for this extension to the
         :attr:`~pystac.STACObject.stac_extensions` list for the given object, if it is
         not already present."""
-        if obj.stac_extensions is None:
-            obj.stac_extensions = [cls.get_schema_uri()]
-        elif not cls.has_extension(obj):
-            obj.stac_extensions.append(cls.get_schema_uri())
+        if isinstance(obj, pystac.Asset):
+            if obj._stac_extensions is None:
+                obj._stac_extensions = [cls.get_schema_uri()]
+            elif not cls.has_extension(obj):
+                obj._stac_extensions.append(cls.get_schema_uri())
+        else:
+            if obj.stac_extensions is None:
+                obj.stac_extensions = [cls.get_schema_uri()]
+            elif not cls.has_extension(obj):
+                obj.stac_extensions.append(cls.get_schema_uri())
 
     @classmethod
     def remove_from(cls, obj: S) -> None:
         """Remove the schema URI for this extension from the
         :attr:`pystac.STACObject.stac_extensions` list for the given object."""
+
         if obj.stac_extensions is not None:
-            obj.stac_extensions = [
-                uri for uri in obj.stac_extensions if uri != cls.get_schema_uri()
-            ]
+            if isinstance(obj, pystac.Asset):
+                obj._stac_extensions = [
+                    uri for uri in obj._stac_extensions if uri != cls.get_schema_uri()
+                ]
+            else:
+                obj.stac_extensions = [
+                    uri for uri in obj.stac_extensions if uri != cls.get_schema_uri()
+                ]
 
     @classmethod
     def has_extension(cls, obj: S) -> bool:
         """Check if the given object implements this extension by checking
         :attr:`pystac.STACObject.stac_extensions` for this extension's schema URI."""
         schema_startswith = VERSION_REGEX.split(cls.get_schema_uri())[0] + "/"
 
+        if isinstance(obj, (pystac.Item, pystac.Collection)):
+            for asset in obj.assets.values():
+                if asset._stac_extensions is not None and any(
+                    uri.startswith(schema_startswith)
+                    for uri in asset._stac_extensions
+                ):
+                    return True
+
+        elif isinstance(obj, pystac.Asset):
+            if obj.owner and obj.owner.stac_extensions is not None and any(
+                uri.startswith(schema_startswith)
+                for uri in obj.owner.stac_extensions
+            ):
+                return True
+            else:
+                return obj._stac_extensions is not None and any(
+                    uri.startswith(schema_startswith)
+                    for uri in obj._stac_extensions
+                )
+
         return obj.stac_extensions is not None and any(
             uri.startswith(schema_startswith) for uri in obj.stac_extensions
         )
@@ -173,15 +205,13 @@ def validate_owner_has_extension(
             STACError : If ``add_if_missing`` is ``True`` and ``asset.owner`` is
                 ``None``.
         """
-        if asset.owner is None:
-            if add_if_missing:
-                raise pystac.STACError(
-                    "Attempted to use add_if_missing=True for an Asset with no owner. "
-                    "Use Asset.set_owner or set add_if_missing=False."
-                )
-            else:
-                return
-        return cls.ensure_has_extension(cast(S, asset.owner), add_if_missing)
+
+        warnings.warn(
+            "validate_owner_has_extension is deprecated and will be removed in v2.0. "
+            "Use ensure_has_extension instead",
+            DeprecationWarning,
+        )
+        return cls.ensure_has_extension(asset, add_if_missing)
 
     @classmethod
     def validate_has_extension(cls, obj: S, add_if_missing: bool = False) -> None:
@@ -224,7 +254,8 @@ def ensure_has_extension(cls, obj: S, add_if_missing: bool = False) -> None:
 
         if not cls.has_extension(obj):
             raise pystac.ExtensionNotImplemented(
-                f"Could not find extension schema URI {cls.get_schema_uri()} in object."
+                f"Could not find extension schema URI {cls.get_schema_uri()} "
+                "in object."
             )
 
     @classmethod

pystac/extensions/classification.py

@@ -534,7 +534,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ClassificationExtension[T]
             cls.ensure_has_extension(obj, add_if_missing)
             return cast(ClassificationExtension[T], ItemClassificationExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(ClassificationExtension[T], AssetClassificationExtension(obj))
         elif isinstance(obj, item_assets.AssetDefinition):
             cls.ensure_has_extension(

pystac/extensions/datacube.py

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

pystac/extensions/eo.py

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

pystac/extensions/file.py

@@ -224,7 +224,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_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cls(obj)
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/pointcloud.py

@@ -455,7 +455,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> PointcloudExtension[T]:
                 raise pystac.ExtensionTypeError(
                     "Pointcloud extension does not apply to Collection Assets."
                 )
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(PointcloudExtension[T], AssetPointcloudExtension(obj))
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/projection.py

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

pystac/extensions/raster.py

@@ -735,7 +735,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> RasterExtension[T]:
             pystac.ExtensionTypeError : If an invalid object type is passed.
         """
         if isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(RasterExtension[T], AssetRasterExtension(obj))
         elif isinstance(obj, item_assets.AssetDefinition):
             cls.ensure_has_extension(

pystac/extensions/sar.py

@@ -319,7 +319,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> SarExtension[T]:
                 raise pystac.ExtensionTypeError(
                     "SAR extension does not apply to Collection Assets."
                 )
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(SarExtension[T], AssetSarExtension(obj))
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/sat.py

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

pystac/extensions/storage.py

@@ -152,7 +152,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> StorageExtension[T]:
             cls.ensure_has_extension(obj, add_if_missing)
             return cast(StorageExtension[T], ItemStorageExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(StorageExtension[T], AssetStorageExtension(obj))
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/table.py

@@ -158,7 +158,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> TableExtension[T]:
             cls.ensure_has_extension(obj, add_if_missing)
             return cast(TableExtension[T], ItemTableExtension(obj))
         if isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(TableExtension[T], AssetTableExtension(obj))
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/timestamps.py

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

pystac/extensions/view.py

@@ -160,7 +160,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> ViewExtension[T]:
             cls.ensure_has_extension(obj, add_if_missing)
             return cast(ViewExtension[T], ItemViewExtension(obj))
         elif isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(ViewExtension[T], AssetViewExtension(obj))
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/extensions/xarray_assets.py

@@ -60,7 +60,7 @@ def ext(cls, obj: T, add_if_missing: bool = False) -> XarrayAssetsExtension[T]:
             cls.ensure_has_extension(obj, add_if_missing)
             return ItemXarrayAssetsExtension(obj)
         if isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return AssetXarrayAssetsExtension(obj)
         else:
             raise pystac.ExtensionTypeError(cls._ext_error_message(obj))

pystac/item.py

@@ -433,7 +433,22 @@ def to_dict(
         if self.bbox is not None:
             d["bbox"] = self.bbox
 
-        if self.stac_extensions is not None:
+        # we use the fact that in recent Python versions, dict keys are ordered
+        # by default
+        stac_extensions: Optional[Dict[str, None]] = None
+        if self.stac_extensions:
+            stac_extensions = dict.fromkeys(self.stac_extensions)
+
+        for asset in self.assets.values():
+            if stac_extensions and asset._stac_extensions:
+                stac_extensions.update(dict.fromkeys(asset._stac_extensions))
+            elif asset._stac_extensions:
+                stac_extensions = dict.fromkeys(asset._stac_extensions)
+
+        if stac_extensions is not None:
+            d["stac_extensions"] = list(stac_extensions.keys())
+
+        if stac_extensions is not None:
             d["stac_extensions"] = self.stac_extensions
 
         if self.collection_id:

tests/extensions/test_custom.py

@@ -54,7 +54,7 @@ def get_schema_uri(cls) -> str:
     @classmethod
     def ext(cls, obj: T, add_if_missing: bool = False) -> "CustomExtension[T]":
         if isinstance(obj, pystac.Asset):
-            cls.validate_owner_has_extension(obj, add_if_missing)
+            cls.ensure_has_extension(obj, add_if_missing)
             return cast(CustomExtension[T], AssetCustomExtension(obj))
         if isinstance(obj, pystac.Item):
             cls.ensure_has_extension(obj, add_if_missing)
@@ -152,6 +152,14 @@ def test_add_to_catalog(self) -> None:
         catalog_as_dict = catalog.to_dict()
         assert catalog_as_dict["test:prop"] == "foo"
 
+    def test_add_to_asset_no_owner(self) -> None:
+        asset = Asset("http://pystac.test/asset.tif")
+        custom = CustomExtension.ext(asset, add_if_missing=True)
+        assert CustomExtension.has_extension(asset)
+        custom.apply("bar")
+        asset_as_dict = asset.to_dict()
+        assert asset_as_dict["test:prop"] == "bar"
+
     def test_add_to_collection(self) -> None:
         collection = Collection(
             "an-id",