feat(storage): implement object contexts and addressing feedback

- Implement ObjectContexts and ObjectCustomContextPayload.
- Add 'contexts' property to Blob.
- Add 'filter_' parameter to list_blobs.
- Add gRPC conversion for contexts and update_mask generation.
- Fix unit tests to avoid ADC errors.
- Format code with ruff.

Co-authored-by: nidhiii-27 <224584462+nidhiii-27@users.noreply.github.com>

Author: google-labs-jules[bot]

packages/google-cloud-storage/google/cloud/storage/_grpc_conversions.py

@@ -30,21 +30,6 @@
     "event_based_hold": "event_based_hold",
 }
 
-# Map REST property names to GCS V2 Object proto field names.
-_PROPERTY_TO_PROTO_FIELD = {
-    "cacheControl": "cache_control",
-    "contentDisposition": "content_disposition",
-    "contentEncoding": "content_encoding",
-    "contentLanguage": "content_language",
-    "contentType": "content_type",
-    "metadata": "metadata",
-    "eventBasedHold": "event_based_hold",
-    "temporaryHold": "temporary_hold",
-    "kmsKeyName": "kms_key",
-    "customTime": "custom_time",
-    "retention": "retention",
-}
-
 
 def blob_to_proto(blob):
     """Converts a Blob instance to a GCS V2 Object proto message."""
@@ -126,54 +111,22 @@ def blob_to_proto(blob):
     return _storage_v2.Object(**resource_params)
 
 
-def proto_to_blob(proto, blob):
-    """Updates a Blob instance from a GCS V2 Object proto message."""
-    from google.cloud._helpers import _datetime_to_rfc3339
-
-    blob._properties["name"] = proto.name
-    if proto.bucket:
-        # Assuming bucket name is the last part of the resource name
-        blob._properties["bucket"] = proto.bucket.split("/")[-1]
-
-    for rest_prop, proto_field in _PROPERTY_TO_PROTO_FIELD.items():
-        if proto_field in proto:
-            value = getattr(proto, proto_field)
-            if proto_field == "metadata":
-                blob._properties[rest_prop] = dict(value)
-            elif proto_field == "custom_time":
-                blob._properties[rest_prop] = _datetime_to_rfc3339(value)
-            elif proto_field == "retention":
-                retention = {"mode": _storage_v2.Object.Retention.Mode.Name(value.mode)}
-                if "retain_until_time" in value:
-                    retention["retainUntilTime"] = _datetime_to_rfc3339(
-                        value.retain_until_time
-                    )
-                blob._properties[rest_prop] = retention
-            else:
-                blob._properties[rest_prop] = value
-
-    if proto.acl:
-        acl_entries = []
-        for entry in proto.acl:
-            acl_entries.append({"role": entry.role, "entity": entry.entity})
-        blob._properties["acl"] = acl_entries
-
-    if "contexts" in proto:
-        custom = {}
-        for key, payload_proto in proto.contexts.custom.items():
-            payload = {"value": payload_proto.value}
-            if "create_time" in payload_proto:
-                payload["createTime"] = _datetime_to_rfc3339(payload_proto.create_time)
-            if "update_time" in payload_proto:
-                payload["updateTime"] = _datetime_to_rfc3339(payload_proto.update_time)
-            custom[key] = payload
-        blob._properties["contexts"] = {"custom": custom}
-
-    return blob
-
-
 def get_update_mask(blob, changes):
     """Generates a FieldMask for gRPC update operations."""
+    # Map REST property names to GCS V2 Object proto field names.
+    property_to_proto_field = {
+        "cacheControl": "cache_control",
+        "contentDisposition": "content_disposition",
+        "contentEncoding": "content_encoding",
+        "contentLanguage": "content_language",
+        "contentType": "content_type",
+        "metadata": "metadata",
+        "eventBasedHold": "event_based_hold",
+        "temporaryHold": "temporary_hold",
+        "kmsKeyName": "kms_key",
+        "customTime": "custom_time",
+        "retention": "retention",
+    }
     paths = []
     for change in changes:
         if change == "contexts":
@@ -184,7 +137,7 @@ def get_update_mask(blob, changes):
                 for key in contexts.custom:
                     paths.append(f"contexts.custom.{key}")
         else:
-            proto_field = _PROPERTY_TO_PROTO_FIELD.get(change)
+            proto_field = property_to_proto_field.get(change)
             if proto_field:
                 paths.append(proto_field)
 

packages/google-cloud-storage/google/cloud/storage/blob.py

@@ -5325,6 +5325,7 @@ def retention_expiration_time(self):
         if retention_expiration_time is not None:
             return _rfc3339_nanos_to_datetime(retention_expiration_time)
 
+
 class ObjectCustomContextPayload(dict):
     """Payload for a custom context.
 

packages/google-cloud-storage/tests/unit/test__grpc_conversions.py

@@ -135,12 +135,16 @@ def test_blob_to_proto_retention():
         retain_until_time.timestamp()
     )
 
+
 def test_blob_to_proto_contexts():
-    blob = mock.Mock(spec=["name", "bucket", "contexts", "custom_time", "acl", "retention"])
+    blob = mock.Mock(
+        spec=["name", "bucket", "contexts", "custom_time", "acl", "retention"]
+    )
     blob.name = "blob-name"
     blob.bucket.name = "bucket-name"
 
     from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
+
     create_time = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
     payload = ObjectCustomContextPayload(value="val", create_time=create_time)
     blob.contexts = ObjectContexts(blob, custom={"key": payload})
@@ -160,41 +164,18 @@ def test_blob_to_proto_contexts():
     )
 
 
-def test_proto_to_blob_contexts():
-    from google.cloud.storage.blob import Blob
-    bucket = mock.Mock()
-    blob = Blob("blob-name", bucket=bucket)
-
-    from google.protobuf import timestamp_pb2
-    create_time = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
-    create_time_proto = timestamp_pb2.Timestamp()
-    create_time_proto.FromDatetime(create_time)
-
-    proto = _storage_v2.Object(
-        name="blob-name",
-        contexts=_storage_v2.ObjectContexts(
-            custom={
-                "key": _storage_v2.ObjectCustomContextPayload(
-                    value="val", create_time=create_time_proto
-                )
-            }
-        )
-    )
-
-    _grpc_conversions.proto_to_blob(proto, blob)
-
-    assert "contexts" in blob._properties
-    assert "key" in blob._properties["contexts"]["custom"]
-    assert blob._properties["contexts"]["custom"]["key"]["value"] == "val"
-    assert blob.contexts.custom["key"].create_time == create_time
-
-
 def test_get_update_mask_contexts():
     blob = mock.Mock(spec=["contexts"])
     from google.cloud.storage.blob import ObjectContexts, ObjectCustomContextPayload
 
     # Partial updates
-    blob.contexts = ObjectContexts(blob, custom={"k1": ObjectCustomContextPayload(value="v1"), "k2": ObjectCustomContextPayload(value="v2")})
+    blob.contexts = ObjectContexts(
+        blob,
+        custom={
+            "k1": ObjectCustomContextPayload(value="v1"),
+            "k2": ObjectCustomContextPayload(value="v2"),
+        },
+    )
     mask = _grpc_conversions.get_update_mask(blob, ["contexts"])
     assert "contexts.custom.k1" in mask.paths
     assert "contexts.custom.k2" in mask.paths
@@ -207,7 +188,9 @@ def test_get_update_mask_contexts():
     assert len(mask.paths) == 1
 
     # Mixed
-    mask = _grpc_conversions.get_update_mask(blob, ["contexts", "metadata", "customTime"])
+    mask = _grpc_conversions.get_update_mask(
+        blob, ["contexts", "metadata", "customTime"]
+    )
     assert "contexts.custom" in mask.paths
     assert "metadata" in mask.paths
     assert "custom_time" in mask.paths

packages/google-cloud-storage/tests/unit/test_blob.py

@@ -6385,12 +6385,14 @@ def delete_blob(
             )
         )
 
+
 import unittest
 import datetime
 import mock
 from google.cloud.storage.blob import Blob, ObjectContexts, ObjectCustomContextPayload
 from google.cloud.storage._helpers import _UTC
 
+
 class TestObjectContexts(unittest.TestCase):
     def test_payload_ctor(self):
         create_time = datetime.datetime(2025, 1, 1, tzinfo=_UTC)
@@ -6425,13 +6427,19 @@ def test_contexts_from_api_repr(self):
         self.assertIn("key", contexts.custom)
         payload = contexts.custom["key"]
         self.assertEqual(payload.value, "val")
-        self.assertEqual(payload.create_time, datetime.datetime(2025, 1, 1, tzinfo=_UTC))
-        self.assertEqual(payload.update_time, datetime.datetime(2025, 1, 2, tzinfo=_UTC))
+        self.assertEqual(
+            payload.create_time, datetime.datetime(2025, 1, 1, tzinfo=_UTC)
+        )
+        self.assertEqual(
+            payload.update_time, datetime.datetime(2025, 1, 2, tzinfo=_UTC)
+        )
 
     def test_blob_contexts_property(self):
         bucket = mock.Mock()
         bucket.name = "b"
-        bucket.__getitem__ = mock.Mock(side_effect=lambda x: "b" if x in (0, -1) else None)
+        bucket.__getitem__ = mock.Mock(
+            side_effect=lambda x: "b" if x in (0, -1) else None
+        )
         blob = Blob("blob-name", bucket=bucket)
         self.assertIsInstance(blob.contexts, ObjectContexts)
         self.assertEqual(blob.contexts.custom, {})
@@ -6443,27 +6451,39 @@ def test_blob_contexts_property(self):
         blob.contexts = None
         self.assertIsNone(blob._properties["contexts"])
 
+
 class TestListBlobsFilter(unittest.TestCase):
-    def test_client_list_blobs_filter(self):
+    @staticmethod
+    def _make_client(*args, **kw):
         from google.cloud.storage.client import Client
+        import google.auth.credentials
+
+        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials.universe_domain = "googleapis.com"
+        kw["credentials"] = kw.get("credentials") or credentials
+        return Client(*args, **kw)
+
+    def test_client_list_blobs_filter(self):
         from google.cloud.storage.bucket import Bucket
-        client = Client(project="p")
+
+        client = self._make_client(project="p")
         bucket = Bucket(client, name="b")
 
         with mock.patch.object(client, "_list_resource") as mocked:
-            list(client.list_blobs(bucket, filter_="contexts.custom.foo = \"bar\""))
+            list(client.list_blobs(bucket, filter_='contexts.custom.foo = "bar"'))
 
             mocked.assert_called_once()
             args, kwargs = mocked.call_args
             extra_params = kwargs["extra_params"]
-            self.assertEqual(extra_params["filter"], "contexts.custom.foo = \"bar\"")
+            self.assertEqual(extra_params["filter"], 'contexts.custom.foo = "bar"')
 
     def test_bucket_list_blobs_filter(self):
         from google.cloud.storage.bucket import Bucket
+
         client = mock.Mock()
         bucket = Bucket(client, name="b")
 
-        bucket.list_blobs(filter_="contexts.custom.foo = \"bar\"")
+        bucket.list_blobs(filter_='contexts.custom.foo = "bar"')
         client.list_blobs.assert_called_with(
             bucket,
             max_results=None,
@@ -6482,14 +6502,25 @@ def test_bucket_list_blobs_filter(self):
             match_glob=None,
             include_folders_as_prefixes=None,
             soft_deleted=None,
-            filter_="contexts.custom.foo = \"bar\"",
+            filter_='contexts.custom.foo = "bar"',
         )
 
+
 class TestSerialization(unittest.TestCase):
-    def test_blob_patch_contexts(self):
+    @staticmethod
+    def _make_client(*args, **kw):
         from google.cloud.storage.client import Client
+        import google.auth.credentials
+
+        credentials = mock.Mock(spec=google.auth.credentials.Credentials)
+        credentials.universe_domain = "googleapis.com"
+        kw["credentials"] = kw.get("credentials") or credentials
+        return Client(*args, **kw)
+
+    def test_blob_patch_contexts(self):
         from google.cloud.storage.bucket import Bucket
-        client = Client(project="p")
+
+        client = self._make_client(project="p")
         bucket = Bucket(client, name="b")
         blob = Blob("blob-name", bucket=bucket)
 
@@ -6504,9 +6535,9 @@ def test_blob_patch_contexts(self):
             self.assertEqual(sent_resource["contexts"]["custom"]["key"]["value"], "val")
 
     def test_blob_patch_contexts_none(self):
-        from google.cloud.storage.client import Client
         from google.cloud.storage.bucket import Bucket
-        client = Client(project="p")
+
+        client = self._make_client(project="p")
         bucket = Bucket(client, name="b")
         blob = Blob("blob-name", bucket=bucket)