RFC: Use mutable dictionaries for object properties

pystac/asset.py

@@ -1,7 +1,8 @@
 from copy import copy
 from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 
-from pystac import common_metadata
+from pystac import base, common_metadata
+from pystac.media_type import MediaType
 from pystac import utils
 
 if TYPE_CHECKING:
@@ -10,7 +11,7 @@
     from pystac.item import Item as Item_Type
 
 
-class Asset:
+class Asset(base.JSONObject):
     """An object that contains a link to data associated with an Item or Collection that
     can be downloaded or streamed.
 
@@ -30,33 +31,13 @@ class Asset:
             object JSON.
     """
 
-    href: str
-    """Link to the asset object. Relative and absolute links are both allowed."""
-
-    title: Optional[str]
-    """Optional displayed title for clients and users."""
-
-    description: Optional[str]
-    """A description of the Asset providing additional details, such as how it was
-    processed or created. CommonMark 0.29 syntax MAY be used for rich text
-    representation."""
-
-    media_type: Optional[str]
-    """Optional description of the media type. Registered Media Types are preferred.
-    See :class:`~pystac.MediaType` for common media types."""
-
-    roles: Optional[List[str]]
-    """Optional, Semantic roles (i.e. thumbnail, overview, data, metadata) of the
-    asset."""
+    fields: Dict[str, Any]
+    """Dictionary of JSON fields for this asset."""
 
     owner: Optional[Union["Item_Type", "Collection_Type"]]
     """The :class:`~pystac.Item` or :class:`~pystac.Collection` that this asset belongs
     to, or ``None`` if it has no owner."""
 
-    extra_fields: Dict[str, Any]
-    """Optional, additional fields for this asset. This is used by extensions as a
-    way to serialize and deserialize properties on asset object JSON."""
-
     def __init__(
         self,
         href: str,
@@ -66,16 +47,78 @@ def __init__(
         roles: Optional[List[str]] = None,
         extra_fields: Optional[Dict[str, Any]] = None,
     ) -> None:
+        self.fields = extra_fields or {}
         self.href = href
         self.title = title
         self.description = description
         self.media_type = media_type
         self.roles = roles
-        self.extra_fields = extra_fields or {}
 
         # The Item which owns this Asset.
         self.owner = None
 
+    @property
+    def href(self) -> str:
+        """Link to the asset object. Relative and absolute links are both allowed."""
+        return utils.get_required(self.fields.get("href"), self, "href")
+
+    @href.setter
+    def href(self, v: str) -> None:
+        self._set_field("href", v)
+
+    @property
+    def title(self) -> Optional[str]:
+        """Optional displayed title for clients and users."""
+        return self.fields.get("title")
+
+    @title.setter
+    def title(self, v: Optional[str]) -> None:
+        self._set_field("title", v, pop_if_none=True)
+
+    @property
+    def description(self) -> Optional[str]:
+        """A description of the Asset providing additional details, such as how it was
+        processed or created. CommonMark 0.29 syntax MAY be used for rich text
+        representation."""
+        return self.fields.get("description")
+
+    @description.setter
+    def description(self, v: Optional[str]) -> None:
+        self._set_field("description", v, pop_if_none=True)
+
+    @property
+    def media_type(self) -> Optional[Union[MediaType, str]]:
+        """Optional description of the media type. Registered Media Types are preferred.
+        See :class:`~pystac.MediaType` for common media types."""
+        return self.fields.get("type")
+
+    @media_type.setter
+    def media_type(self, v: Optional[Union[MediaType, str]]) -> None:
+        self._set_field("type", v, pop_if_none=True)
+
+    @property
+    def roles(self) -> Optional[List[str]]:
+        """Optional, Semantic roles (i.e. thumbnail, overview, data, metadata) of the
+        asset."""
+        return self.fields.get("roles")
+
+    @roles.setter
+    def roles(self, v: Optional[List[str]]) -> None:
+        self._set_field("roles", v, pop_if_none=True)
+
+    @property
+    def extra_fields(self) -> Dict[str, Any]:
+        return {
+            k: v
+            for k, v in self.fields.items()
+            if k not in {"href", "title", "description", "type", "roles"}
+        }
+
+    @extra_fields.setter
+    def extra_fields(self, v: Dict[str, Any]) -> None:
+        for _k, _v in v.items():
+            self._set_field(_k, _v, pop_if_none=True)
+
     def set_owner(self, obj: Union["Collection_Type", "Item_Type"]) -> None:
         """Sets the owning item of this Asset.
 
@@ -104,33 +147,6 @@ def get_absolute_href(self) -> Optional[str]:
             else:
                 return None
 
-    def to_dict(self) -> Dict[str, Any]:
-        """Generate a dictionary representing the JSON of this Asset.
-
-        Returns:
-            dict: A serialization of the Asset that can be written out as JSON.
-        """
-
-        d: Dict[str, Any] = {"href": self.href}
-
-        if self.media_type is not None:
-            d["type"] = self.media_type
-
-        if self.title is not None:
-            d["title"] = self.title
-
-        if self.description is not None:
-            d["description"] = self.description
-
-        if self.extra_fields is not None and len(self.extra_fields) > 0:
-            for k, v in self.extra_fields.items():
-                d[k] = v
-
-        if self.roles is not None:
-            d["roles"] = self.roles
-
-        return d
-
     def clone(self) -> "Asset":
         """Clones this asset.
 
@@ -144,7 +160,7 @@ def clone(self) -> "Asset":
             description=self.description,
             media_type=self.media_type,
             roles=self.roles,
-            extra_fields=self.extra_fields,
+            extra_fields=dict(self.extra_fields),
         )
 
     @property

pystac/base.py

@@ -0,0 +1,28 @@
+from copy import deepcopy
+from typing import Any, Dict, TypeVar
+
+
+V = TypeVar("V")
+
+
+class JSONObject:
+    """A base class for objects that can be represented as JSON objects.
+
+    Instances of ``JSONObject`` have a :attr:`JSONObject.fields` attribute that is a
+    dictionary representation of the JSON object. Any property getters/setters on
+    inheriting classes MUST be sure to keep this dictionary up-to-date.
+    """
+
+    fields: Dict[str, Any] = {}
+
+    def to_dict(self, preserve_dict: bool = True) -> Dict[str, Any]:
+        if preserve_dict:
+            return deepcopy(self.fields)
+        else:
+            return self.fields
+
+    def _set_field(self, name: str, v: Any, *, pop_if_none: bool = False) -> None:
+        if v is None and pop_if_none:
+            self.fields.pop(name, None)
+        else:
+            self.fields[name] = v

pystac/common_metadata.py

@@ -1,9 +1,10 @@
 from datetime import datetime as Datetime
-from pystac.errors import STACError
 from typing import Any, cast, Dict, List, Optional, Type, TYPE_CHECKING, TypeVar, Union
 
 import pystac
 from pystac import utils
+from pystac.base import JSONObject
+from pystac.errors import STACError
 
 if TYPE_CHECKING:
     from pystac.provider import Provider as Provider_Type
@@ -32,19 +33,16 @@ def __init__(self, object: Union["Asset_Type", "Item_Type"]):
     def _set_field(self, prop_name: str, v: Optional[Any]) -> None:
         if hasattr(self.object, prop_name):
             setattr(self.object, prop_name, v)
-        elif hasattr(self.object, "properties"):
-            item = cast("Item_Type", self.object)
+        elif isinstance(self.object, pystac.Item):
             if v is None:
-                item.properties.pop(prop_name, None)
+                self.object.properties.pop(prop_name, None)
             else:
-                item.properties[prop_name] = v
-        elif hasattr(self.object, "extra_fields") and isinstance(
-            self.object.extra_fields, Dict
-        ):
+                self.object.properties[prop_name] = v
+        elif isinstance(self.object, JSONObject):
             if v is None:
-                self.object.extra_fields.pop(prop_name, None)
+                self.object.fields.pop(prop_name, None)
             else:
-                self.object.extra_fields[prop_name] = v
+                self.object.fields[prop_name] = v
         else:
             raise pystac.STACError(f"Cannot set field {prop_name} on {self}.")
 

pystac/extensions/datacube.py

@@ -528,7 +528,7 @@ class AssetDatacubeExtension(DatacubeExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
         else:

pystac/extensions/eo.py

@@ -455,7 +455,7 @@ def _get_bands(self) -> Optional[List[Band]]:
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/file.py

@@ -102,7 +102,7 @@ class FileExtension(
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/pointcloud.py

@@ -495,7 +495,7 @@ class AssetPointcloudExtension(PointcloudExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
             self.repr_id = f"href={asset.href} item.id={asset.owner.id}"

pystac/extensions/projection.py

@@ -335,7 +335,7 @@ class AssetProjectionExtension(ProjectionExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/raster.py

@@ -651,7 +651,7 @@ class RasterExtension(
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/sar.py

@@ -381,7 +381,7 @@ class AssetSarExtension(SarExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/sat.py

@@ -214,7 +214,7 @@ class AssetSatExtension(SatExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/timestamps.py

@@ -192,7 +192,7 @@ class AssetTimestampsExtension(TimestampsExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/extensions/view.py

@@ -221,7 +221,7 @@ class AssetViewExtension(ViewExtension[pystac.Asset]):
 
     def __init__(self, asset: pystac.Asset):
         self.asset_href = asset.href
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner and isinstance(asset.owner, pystac.Item):
             self.additional_read_properties = [asset.owner.properties]
 

pystac/item.py

@@ -168,10 +168,10 @@ def get_datetime(self, asset: Optional[Asset] = None) -> Optional[Datetime]:
         Returns:
             datetime or None
         """
-        if asset is None or "datetime" not in asset.extra_fields:
+        if asset is None or "datetime" not in asset.fields:
             return self.datetime
         else:
-            asset_dt = asset.extra_fields.get("datetime")
+            asset_dt = asset.fields.get("datetime")
             if asset_dt is None:
                 return None
             else:
@@ -186,7 +186,7 @@ def set_datetime(self, datetime: Datetime, asset: Optional[Asset] = None) -> Non
         if asset is None:
             self.datetime = datetime
         else:
-            asset.extra_fields["datetime"] = datetime_to_str(datetime)
+            asset.fields["datetime"] = datetime_to_str(datetime)
 
     def get_assets(self) -> Dict[str, Asset]:
         """Get this item's assets.

tests/extensions/test_custom.py

@@ -94,7 +94,7 @@ def __init__(self, item: pystac.Item) -> None:
 class AssetCustomExtension(CustomExtension[pystac.Asset]):
     def __init__(self, asset: pystac.Asset) -> None:
         self.catalog = asset
-        self.properties = asset.extra_fields
+        self.properties = asset.fields
         if asset.owner:
             if isinstance(asset.owner, pystac.Item):
                 self.additional_read_properties = [asset.owner.properties]

tests/extensions/test_pointcloud.py

@@ -271,7 +271,7 @@ def test_asset_extension(self) -> None:
         pc_item.add_asset("data", asset)
         ext = AssetPointcloudExtension(asset)
         self.assertEqual(ext.asset_href, asset.href)
-        self.assertEqual(ext.properties, asset.extra_fields)
+        self.assertEqual(ext.properties, asset.fields)
         self.assertEqual(ext.additional_read_properties, [pc_item.properties])
 
     def test_ext(self) -> None: