Preserve Collection assets on clone (#834)

* Allow assets to be passed to Item init

* Allow assets to be passed to Collection init

* Preserve Collection assets on clone

* Add CHANGELOG entry for #834

* Match clone tests for Item and Collection

CHANGELOG.md

@@ -11,6 +11,7 @@
 - Updated AssetDefinition to have create, apply methods ([#768](https://github.com/stac-utils/pystac/pull/768))
 - Add Grid Extension support ([#799](https://github.com/stac-utils/pystac/pull/799))
 - Rich HTML representations for Jupyter Notebook display ([#743](https://github.com/stac-utils/pystac/pull/743))
+- Add `assets` argument to `Item` and `Collection` init methods to allow adding Assets during object initialization ([#834](https://github.com/stac-utils/pystac/pull/834))
 
 ### Removed
 
@@ -24,6 +25,7 @@
 - "How to create STAC catalogs" tutorial ([#775](https://github.com/stac-utils/pystac/pull/775))
 - Add a `variables` argument, to accompany `dimensions`, for the `apply` method of stac objects extended with datacube ([#782](https://github.com/stac-utils/pystac/pull/782))
 - Deepcopy collection properties on clone. Implement `clone` method for `Summaries` ([#794](https://github.com/stac-utils/pystac/pull/794))
+- Collection assets are now preserved when using `Collection.clone` ([#834](https://github.com/stac-utils/pystac/pull/834))
 - Docstrings for `StacIO.read_text` and `StacIO.write_text` now match the type annotations for the `source` argument. ([#835](https://github.com/stac-utils/pystac/pull/835))
 
 ## [v1.4.0]

pystac/collection.py

@@ -1,6 +1,7 @@
 from html import escape
 from copy import deepcopy
 from datetime import datetime
+
 from pystac.errors import STACTypeError
 from pystac.html.jinja_env import get_jinja_env
 from typing import (
@@ -446,6 +447,9 @@ class Collection(Catalog):
             either a set of values or statistics such as a range.
         extra_fields : Extra fields that are part of the top-level
             JSON properties of the Collection.
+        assets : A dictionary mapping string keys to :class:`~pystac.Asset` objects. All
+            :class:`~pystac.Asset` values in the dictionary will have their
+            :attr:`~pystac.Asset.owner` attribute set to the created Collection.
     """
 
     assets: Dict[str, Asset]
@@ -504,6 +508,7 @@ def __init__(
         keywords: Optional[List[str]] = None,
         providers: Optional[List["Provider_Type"]] = None,
         summaries: Optional[Summaries] = None,
+        assets: Optional[Dict[str, Asset]] = None,
     ):
         super().__init__(
             id,
@@ -523,6 +528,9 @@ def __init__(
         self.summaries = summaries or Summaries.empty()
 
         self.assets = {}
+        if assets is not None:
+            for k, asset in assets.items():
+                self.add_asset(k, asset)
 
     def __repr__(self) -> str:
         return "<Collection id={}>".format(self.id)
@@ -579,6 +587,7 @@ def clone(self) -> "Collection":
             keywords=self.keywords.copy() if self.keywords is not None else None,
             providers=deepcopy(self.providers),
             summaries=self.summaries.clone(),
+            assets={k: asset.clone() for k, asset in self.assets.items()},
         )
 
         clone._resolved_objects.cache(clone)
@@ -631,7 +640,9 @@ def from_dict(
         if summaries is not None:
             summaries = Summaries(summaries)
 
-        assets: Optional[Dict[str, Any]] = d.get("assets", None)
+        assets: Optional[Dict[str, Any]] = {
+            k: Asset.from_dict(v) for k, v in d.get("assets", {}).items()
+        }
         links = d.pop("links")
 
         d.pop("stac_version")
@@ -649,6 +660,7 @@ def from_dict(
             summaries=summaries,
             href=href,
             catalog_type=catalog_type,
+            assets=assets,
         )
 
         for link in links:
@@ -659,10 +671,6 @@ def from_dict(
             if link["rel"] != pystac.RelType.SELF or href is None:
                 collection.add_link(Link.from_dict(link))
 
-        if assets is not None:
-            for asset_key, asset_dict in assets.items():
-                collection.add_asset(asset_key, Asset.from_dict(asset_dict))
-
         if root:
             collection.set_root(root)
 

pystac/item.py

@@ -51,6 +51,9 @@ class Item(STACObject):
             belongs to.
         extra_fields : Extra fields that are part of the top-level JSON
             properties of the Item.
+        assets : A dictionary mapping string keys to :class:`~pystac.Asset` objects. All
+            :class:`~pystac.Asset` values in the dictionary will have their
+            :attr:`~pystac.Asset.owner` attribute set to the created Item.
     """
 
     assets: Dict[str, Asset]
@@ -107,6 +110,7 @@ def __init__(
         href: Optional[str] = None,
         collection: Optional[Union[str, Collection]] = None,
         extra_fields: Optional[Dict[str, Any]] = None,
+        assets: Optional[Dict[str, Asset]] = None,
     ):
         super().__init__(stac_extensions or [])
 
@@ -144,6 +148,11 @@ def __init__(
             else:
                 self.collection_id = collection
 
+        self.assets = {}
+        if assets is not None:
+            for k, asset in assets.items():
+                self.add_asset(k, asset)
+
     def __repr__(self) -> str:
         return "<Item id={}>".format(self.id)
 
@@ -359,13 +368,11 @@ def clone(self) -> "Item":
             properties=deepcopy(self.properties),
             stac_extensions=deepcopy(self.stac_extensions),
             collection=self.collection_id,
+            assets={k: asset.clone() for k, asset in self.assets.items()},
         )
         for link in self.links:
             clone.add_link(link.clone())
 
-        for k, asset in self.assets.items():
-            clone.add_asset(k, asset.clone())
-
         return clone
 
     def _object_links(self) -> List[Union[str, pystac.RelType]]:
@@ -420,6 +427,7 @@ def from_dict(
             stac_extensions=stac_extensions,
             collection=collection_id,
             extra_fields=d,
+            assets={k: Asset.from_dict(v) for k, v in assets.items()},
         )
 
         has_self_link = False
@@ -430,11 +438,6 @@ def from_dict(
         if not has_self_link and href is not None:
             item.add_link(Link.self_href(href))
 
-        for k, v in assets.items():
-            asset = Asset.from_dict(v)
-            asset.set_owner(item)
-            item.assets[k] = asset
-
         if root:
             item.set_root(root)
 

tests/test_collection.py

@@ -271,6 +271,25 @@ def test_from_invalid_dict_raises_exception(self) -> None:
         with self.assertRaises(pystac.STACTypeError):
             _ = pystac.Collection.from_dict(catalog_dict)
 
+    def test_clone_preserves_assets(self) -> None:
+        path = TestCases.get_path("data-files/collections/with-assets.json")
+        original_collection = Collection.from_file(path)
+        assert len(original_collection.assets) > 0
+        assert all(
+            asset.owner is original_collection
+            for asset in original_collection.assets.values()
+        )
+
+        cloned_collection = original_collection.clone()
+
+        for key in original_collection.assets:
+            with self.subTest(f"Preserves {key} asset"):
+                self.assertIn(key, cloned_collection.assets)
+            cloned_asset = cloned_collection.assets.get(key)
+            if cloned_asset is not None:
+                with self.subTest(f"Sets owner for {key}"):
+                    self.assertIs(cloned_asset.owner, cloned_collection)
+
 
 class ExtentTest(unittest.TestCase):
     def setUp(self) -> None:

tests/test_item.py

@@ -221,15 +221,23 @@ def test_0_9_item_with_no_extensions_does_not_read_collection_data(self) -> None
         )
         self.assertFalse(did_merge)
 
-    def test_clone_sets_asset_owner(self) -> None:
+    def test_clone_preserves_assets(self) -> None:
         cat = TestCases.test_case_2()
-        item = next(iter(cat.get_all_items()))
-        original_asset = list(item.assets.values())[0]
-        assert original_asset.owner is item
+        original_item = next(iter(cat.get_all_items()))
+        assert len(original_item.assets) > 0
+        assert all(
+            asset.owner is original_item for asset in original_item.assets.values()
+        )
+
+        cloned_item = original_item.clone()
 
-        clone = item.clone()
-        clone_asset = list(clone.assets.values())[0]
-        self.assertIs(clone_asset.owner, clone)
+        for key in original_item.assets:
+            with self.subTest(f"Preserves {key} asset"):
+                self.assertIn(key, cloned_item.assets)
+            cloned_asset = cloned_item.assets.get(key)
+            if cloned_asset is not None:
+                with self.subTest(f"Sets owner for {key}"):
+                    self.assertIs(cloned_asset.owner, cloned_item)
 
     def test_make_asset_href_relative_is_noop_on_relative_hrefs(self) -> None:
         cat = TestCases.test_case_2()