feat: add ObjectContexts to Blob

google/cloud/storage/blob.py

@@ -75,6 +75,9 @@
 from google.cloud.storage.exceptions import DataCorruption
 from google.cloud.storage.exceptions import InvalidResponse
 from google.cloud.storage.retry import ConditionalRetryPolicy
+
+from google.cloud.storage.contexts import ObjectContexts
+from google.cloud.storage.contexts import ObjectCustomContextPayload
 from google.cloud.storage.retry import DEFAULT_RETRY
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
@@ -4765,6 +4768,86 @@ def media_link(self):
         """
         return self._properties.get("mediaLink")
 
+
+    @property
+    def contexts(self):
+        """Retrieve or set the user-defined object contexts for this blob.
+
+        See https://cloud.google.com/storage/docs/json_api/v1/objects
+
+        .. note::
+            The getter for this property returns an
+            :class:`~google.cloud.storage.contexts.ObjectContexts` object, which is a
+            structured representation of the blob's contexts configuration.
+            Modifying the returned object has no effect. To update the blob's
+            contexts, create and assign a new ``ObjectContexts`` object to this
+            property and then call
+            :meth:`~google.cloud.storage.blob.Blob.patch`.
+
+            .. code-block:: python
+
+                from google.cloud.storage.contexts import (
+                    ObjectContexts,
+                    ObjectCustomContextPayload,
+                )
+
+                contexts = ObjectContexts()
+                contexts.custom = {"foo": ObjectCustomContextPayload(value="Bar")}
+                blob.contexts = contexts
+                blob.patch()
+
+        :setter: Set the user-defined object contexts for this blob.
+        :getter: Gets the user-defined object contexts for this blob.
+
+        :rtype: :class:`~google.cloud.storage.contexts.ObjectContexts` or ``NoneType``
+        :returns:
+            An ``ObjectContexts`` object representing the configuration, or ``None``
+            if no contexts are configured.
+        """
+        resource = self._properties.get("contexts")
+        if resource:
+            return ObjectContexts._from_api_resource(resource)
+        return None
+
+    @contexts.setter
+    def contexts(self, value):
+
+        if value is None:
+            self._patch_property("contexts", None)
+        elif isinstance(value, ObjectContexts):
+            self._patch_property("contexts", value._to_api_resource())
+        elif isinstance(value, dict):
+            # Parse dict using ObjectContexts helpers to be tolerant.
+            if "custom" in value:
+                custom_payloads = {}
+                for k, v in value["custom"].items():
+                    if v is None:
+                        custom_payloads[k] = None
+                    elif isinstance(v, ObjectCustomContextPayload):
+                        custom_payloads[k] = v
+                    elif isinstance(v, dict):
+                        custom_payloads[k] = ObjectCustomContextPayload(value=v.get("value"))
+                    else:
+                        custom_payloads[k] = ObjectCustomContextPayload(value=str(v))
+                patch_value = ObjectContexts(custom=custom_payloads)._to_api_resource()
+                self._patch_property("contexts", patch_value)
+            else:
+                # Flat dict {"foo": "bar"}
+                custom_payloads = {}
+                for k, v in value.items():
+                    if v is None:
+                        custom_payloads[k] = None
+                    elif isinstance(v, ObjectCustomContextPayload):
+                        custom_payloads[k] = v
+                    elif isinstance(v, dict):
+                        custom_payloads[k] = ObjectCustomContextPayload(value=v.get("value"))
+                    else:
+                        custom_payloads[k] = ObjectCustomContextPayload(value=str(v))
+                patch_value = ObjectContexts(custom=custom_payloads)._to_api_resource()
+                self._patch_property("contexts", patch_value)
+        else:
+            self._patch_property("contexts", value)
+
     @property
     def metadata(self):
         """Retrieve arbitrary/application specific metadata for the object.

google/cloud/storage/contexts.py

@@ -0,0 +1,130 @@
+# Copyright 2024 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""User-defined object contexts for Google Cloud Storage."""
+
+from typing import Dict, Any, Optional
+import datetime
+
+from google.cloud._helpers import _rfc3339_nanos_to_datetime
+from google.cloud._helpers import _datetime_to_rfc3339
+
+_VALUE = "value"
+_CREATE_TIME = "createTime"
+_UPDATE_TIME = "updateTime"
+_CUSTOM = "custom"
+
+
+class ObjectCustomContextPayload:
+    """The payload of a single user-defined object context.
+
+    :type value: str
+    :param value: The value of the object context.
+
+    :type create_time: :class:`datetime.datetime` or ``NoneType``
+    :param create_time: (Optional) The time at which the object context was created.
+
+    :type update_time: :class:`datetime.datetime` or ``NoneType``
+    :param update_time: (Optional) The time at which the object context was last updated.
+    """
+
+    def __init__(
+        self,
+        value: str,
+        create_time: Optional[datetime.datetime] = None,
+        update_time: Optional[datetime.datetime] = None,
+    ):
+        self.value = value
+        self.create_time = create_time
+        self.update_time = update_time
+
+    @classmethod
+    def _from_api_resource(cls, resource: Dict[str, Any]) -> "ObjectCustomContextPayload":
+        """Factory: creates an ObjectCustomContextPayload instance from a server response."""
+        create_time = resource.get(_CREATE_TIME)
+        if create_time:
+            create_time = _rfc3339_nanos_to_datetime(create_time)
+
+        update_time = resource.get(_UPDATE_TIME)
+        if update_time:
+            update_time = _rfc3339_nanos_to_datetime(update_time)
+
+        return cls(
+            value=resource.get(_VALUE),
+            create_time=create_time,
+            update_time=update_time,
+        )
+
+    def _to_api_resource(self) -> Dict[str, Any]:
+        """Serializes this object to a dictionary for API requests."""
+        resource = {_VALUE: self.value}
+        if self.create_time:
+            resource[_CREATE_TIME] = _datetime_to_rfc3339(self.create_time)
+        if self.update_time:
+            resource[_UPDATE_TIME] = _datetime_to_rfc3339(self.update_time)
+        return resource
+
+
+class ObjectContexts:
+    """User-defined object contexts.
+
+    This class is a helper for constructing the contexts dictionary to be
+    assigned to a blob's ``contexts`` property.
+
+    :type custom: dict or ``NoneType``
+    :param custom:
+        (Optional) User-defined object contexts, a dictionary mapping string keys
+        to :class:`ObjectCustomContextPayload` instances. To delete a context via
+        patch, the payload can be mapped to ``None``.
+    """
+
+    def __init__(
+        self,
+        custom: Optional[Dict[str, Optional[ObjectCustomContextPayload]]] = None,
+    ):
+        self.custom = custom or {}
+
+    @classmethod
+    def _from_api_resource(cls, resource: Dict[str, Any]) -> "ObjectContexts":
+        """Factory: creates an ObjectContexts instance from a server response."""
+        custom_data = resource.get(_CUSTOM)
+        custom = {}
+        if custom_data:
+            for key, payload in custom_data.items():
+                if payload is not None:
+                    custom[key] = ObjectCustomContextPayload._from_api_resource(payload)
+                else:
+                    custom[key] = None
+
+        return cls(custom=custom)
+
+    def _to_api_resource(self) -> Dict[str, Any]:
+        """Serializes this object to a dictionary for API requests."""
+        resource = {}
+        if self.custom is not None:
+            custom_resource = {}
+            for key, payload in self.custom.items():
+                if payload is None:
+                    custom_resource[key] = None
+                elif isinstance(payload, ObjectCustomContextPayload):
+                    custom_resource[key] = payload._to_api_resource()
+                elif isinstance(payload, dict):
+                    # We also support a pure dict fallback payload
+                    custom_resource[key] = ObjectCustomContextPayload(value=payload.get(_VALUE))._to_api_resource()
+                elif isinstance(payload, str):
+                    custom_resource[key] = ObjectCustomContextPayload(value=payload)._to_api_resource()
+                else:
+                    custom_resource[key] = ObjectCustomContextPayload(value=str(payload))._to_api_resource()
+            resource[_CUSTOM] = custom_resource
+        return resource

tests/system/test_blob.py

@@ -39,6 +39,50 @@ def _check_blob_hash(blob, info):
     assert md5_hash == info["hash"]
 
 
+
+def test_blob_contexts(shared_bucket, blobs_to_delete):
+    from google.cloud.storage.contexts import ObjectContexts
+    from google.cloud.storage.contexts import ObjectCustomContextPayload
+
+    blob = shared_bucket.blob("test-contexts-blob")
+    blob.upload_from_string(b"Hello, World!")
+
+    # 1. Update/Add custom contexts
+    contexts = ObjectContexts()
+    contexts.custom = {
+        "context_key_1": ObjectCustomContextPayload(value="value_1"),
+        "context_key_2": ObjectCustomContextPayload(value="value_2"),
+    }
+    blob.contexts = contexts
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts is not None
+    assert "context_key_1" in blob.contexts.custom
+    assert blob.contexts.custom["context_key_1"].value == "value_1"
+    assert blob.contexts.custom["context_key_1"].create_time is not None
+    assert blob.contexts.custom["context_key_1"].update_time is not None
+    assert blob.contexts.custom["context_key_2"].value == "value_2"
+
+    # 2. Update existing and Delete one context key
+    blob.contexts = {"context_key_1": "updated_value", "context_key_2": None}
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts is not None
+    assert "context_key_1" in blob.contexts.custom
+    assert blob.contexts.custom["context_key_1"].value == "updated_value"
+    assert "context_key_2" not in blob.contexts.custom
+
+    # 3. Delete all contexts
+    blob.contexts = None
+    blob.patch()
+
+    blob.reload()
+    assert blob.contexts is None
+
+    blobs_to_delete.append(blob)
+
 def test_large_file_write_from_stream_w_user_provided_checksum(
     shared_bucket,
     blobs_to_delete,

tests/unit/test_blob.py

@@ -5712,6 +5712,80 @@ def test_media_link(self):
         blob = self._make_one(BLOB_NAME, bucket=bucket, properties=properties)
         self.assertEqual(blob.media_link, MEDIA_LINK)
 
+
+    def test_contexts_getter(self):
+        from google.cloud.storage.contexts import ObjectContexts
+
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        PROPERTIES = {
+            "contexts": {
+                "custom": {
+                    "foo": {"value": "Foo"},
+                    "bar": {
+                        "value": "Bar",
+                        "createTime": "2023-01-01T12:00:00Z",
+                        "updateTime": "2023-01-02T12:00:00Z",
+                    },
+                }
+            }
+        }
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=PROPERTIES)
+        contexts = blob.contexts
+
+        self.assertIsInstance(contexts, ObjectContexts)
+        self.assertEqual(contexts.custom["foo"].value, "Foo")
+        self.assertIsNone(contexts.custom["foo"].create_time)
+        self.assertEqual(contexts.custom["bar"].value, "Bar")
+        self.assertEqual(
+            contexts.custom["bar"].create_time.isoformat(),
+            "2023-01-01T12:00:00+00:00"
+        )
+        self.assertEqual(
+            contexts.custom["bar"].update_time.isoformat(),
+            "2023-01-02T12:00:00+00:00"
+        )
+
+    def test_contexts_setter_w_dict(self):
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+
+        self.assertIsNone(blob.contexts)
+        blob.contexts = {"foo": "Foo", "bar": None}
+
+        self.assertIn("contexts", blob._changes)
+        self.assertEqual(
+            blob._properties["contexts"],
+            {"custom": {"foo": {"value": "Foo"}, "bar": None}}
+        )
+
+    def test_contexts_setter_w_object_contexts(self):
+        from google.cloud.storage.contexts import ObjectContexts, ObjectCustomContextPayload
+
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+
+        self.assertIsNone(blob.contexts)
+        blob.contexts = ObjectContexts(custom={"foo": ObjectCustomContextPayload(value="Foo")})
+
+        self.assertIn("contexts", blob._changes)
+        self.assertEqual(
+            blob._properties["contexts"],
+            {"custom": {"foo": {"value": "Foo"}}}
+        )
+
+    def test_contexts_setter_none(self):
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        PROPERTIES = {"contexts": {"custom": {"foo": {"value": "Foo"}}}}
+        blob = self._make_one(BLOB_NAME, bucket=bucket, properties=PROPERTIES)
+
+        blob.contexts = None
+        self.assertIn("contexts", blob._changes)
+        self.assertIsNone(blob._properties["contexts"])
+
     def test_metadata_getter(self):
         BLOB_NAME = "blob-name"
         bucket = _Bucket()