feat(storage): add object contexts and list filter support

Implemented the "Object Contexts" feature in the GCS Python SDK.
- Added ObjectContexts class and contexts property to Blob.
- Supported set, get, delete, and clear operations for custom contexts.
- Updated list_blobs in Client and Bucket to support server-side filtering via filter_ parameter.
- Added gRPC conversion support for object contexts.
- Added unit and system tests.
- Addressed code review feedback: renamed internal field to objectContexts, implemented caching for contexts property, and added timestamp support.

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

@@ -87,4 +87,22 @@ def blob_to_proto(blob):
             retain_until_time=retain_until_time_proto,
         )
 
+    contexts = getattr(blob, "contexts", None)
+    if contexts:
+        custom_contexts = contexts.get("custom")
+        if custom_contexts is not None:
+            custom_contexts_proto = {}
+            for key, payload in custom_contexts.items():
+                if payload is not None:
+                    custom_contexts_proto[key] = _storage_v2.ObjectCustomContextPayload(
+                        value=payload.get("value")
+                    )
+
+            resource_params["contexts"] = _storage_v2.ObjectContexts(
+                custom=custom_contexts_proto
+            )
+        else:
+            # Signal clearing of all custom contexts.
+            resource_params["contexts"] = _storage_v2.ObjectContexts(custom=None)
+
     return _storage_v2.Object(**resource_params)

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

@@ -101,7 +101,7 @@
     "crc32c",
     "customTime",
     "md5Hash",
-    "contexts",
+    "objectContexts",
     "metadata",
     "name",
     "retention",
@@ -233,13 +233,23 @@ def __init__(
         )
 
         self._encryption_key = encryption_key
+        self._contexts = None
 
         if kms_key_name is not None:
             self._properties["kmsKeyName"] = kms_key_name
 
         if generation is not None:
             self._properties["generation"] = generation
 
+    def _set_properties(self, value):
+        """Set the properties for the current object.
+
+        :type value: dict or :class:`google.cloud.storage.batch._FutureDict`
+        :param value: The properties to be set.
+        """
+        super()._set_properties(value)
+        self._contexts = None
+
     @property
     def bucket(self):
         """Bucket which contains the object.
@@ -5016,8 +5026,10 @@ def contexts(self):
         :rtype: :class:`ObjectContexts`
         :returns: an instance for managing the object's contexts.
         """
-        info = self._properties.get("contexts", {})
-        return ObjectContexts.from_api_repr(info, self)
+        if self._contexts is None:
+            info = self._properties.get("objectContexts", {})
+            self._contexts = ObjectContexts.from_api_repr(info, self)
+        return self._contexts
 
     @property
     def soft_delete_time(self):
@@ -5343,6 +5355,18 @@ def from_api_repr(cls, resource, blob):
         """
         instance = cls(blob)
         if resource:
+            # Handle timestamps in the resource if present
+            custom = resource.get("custom")
+            if custom:
+                for payload in custom.values():
+                    if payload and "createTime" in payload:
+                        payload["create_time"] = _rfc3339_nanos_to_datetime(
+                            payload["createTime"]
+                        )
+                    if payload and "updateTime" in payload:
+                        payload["update_time"] = _rfc3339_nanos_to_datetime(
+                            payload["updateTime"]
+                        )
             instance.update(resource)
         return instance
 
@@ -5369,7 +5393,7 @@ def set_custom_context(self, key, value):
             custom = {}
             self["custom"] = custom
         custom[key] = {"value": value}
-        self.blob._patch_property("contexts", self)
+        self.blob._patch_property("objectContexts", self)
 
     def delete_custom_context(self, key):
         """Delete a custom context.
@@ -5380,9 +5404,9 @@ def delete_custom_context(self, key):
         custom = self.get("custom")
         if custom is not None:
             custom[key] = None
-            self.blob._patch_property("contexts", self)
+            self.blob._patch_property("objectContexts", self)
 
     def clear_custom_contexts(self):
         """Clear all custom contexts."""
         self["custom"] = None
-        self.blob._patch_property("contexts", self)
+        self.blob._patch_property("objectContexts", self)

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

@@ -1423,6 +1423,8 @@ def list_blobs(
         include_folders_as_prefixes=None,
         soft_deleted=None,
         page_size=None,
+        *,
+        filter_=None,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -1521,6 +1523,12 @@ def list_blobs(
             (Optional) Maximum number of blobs to return in each page.
             Defaults to a value set by the API.
 
+        :type filter_: str
+        :param filter_:
+            (Optional) A filter expression that filters objects listed in the response.
+            The expression must be specified in the GCS filter syntax.
+            See: https://cloud.google.com/storage/docs/json_api/v1/objects/list#filter
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
@@ -1545,6 +1553,7 @@ def list_blobs(
                 match_glob=match_glob,
                 include_folders_as_prefixes=include_folders_as_prefixes,
                 soft_deleted=soft_deleted,
+                filter_=filter_,
             )
 
     def list_notifications(

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

@@ -1291,6 +1291,8 @@ def list_blobs(
         match_glob=None,
         include_folders_as_prefixes=None,
         soft_deleted=None,
+        *,
+        filter_=None,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -1400,6 +1402,11 @@ def list_blobs(
                 Note ``soft_deleted`` and ``versions`` cannot be set to True simultaneously. See:
                 https://cloud.google.com/storage/docs/soft-delete
 
+            filter_ (str):
+                (Optional) A filter expression that filters objects listed in the response.
+                The expression must be specified in the GCS filter syntax.
+                See: https://cloud.google.com/storage/docs/json_api/v1/objects/list#filter
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments. The RPC call
@@ -1440,6 +1447,9 @@ def list_blobs(
             if include_folders_as_prefixes is not None:
                 extra_params["includeFoldersAsPrefixes"] = include_folders_as_prefixes
 
+            if filter_ is not None:
+                extra_params["filter"] = filter_
+
             if soft_deleted is not None:
                 extra_params["softDeleted"] = soft_deleted
 

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

@@ -1209,3 +1209,77 @@ def test_blob_download_as_bytes_single_shot_download(
 
     result_single_shot_download = blob.download_as_bytes(single_shot_download=True)
     assert result_single_shot_download == payload
+
+def test_blob_contexts(shared_bucket, blobs_to_delete):
+    blob_name = f"context-test-{uuid.uuid4().hex}"
+    blob = shared_bucket.blob(blob_name)
+    blob.upload_from_string(b"foo")
+    blobs_to_delete.append(blob)
+
+    # Set context
+    blob.contexts.set_custom_context("foo", "bar")
+    blob.patch()
+
+    assert blob.contexts["custom"]["foo"]["value"] == "bar"
+    assert "create_time" in blob.contexts["custom"]["foo"]
+
+    # Reload and check
+    blob.reload()
+    assert blob.contexts["custom"]["foo"]["value"] == "bar"
+
+    # Update context
+    blob.contexts.set_custom_context("foo", "baz")
+    blob.patch()
+    assert blob.contexts["custom"]["foo"]["value"] == "baz"
+
+    # Add another context
+    blob.contexts.set_custom_context("another", "value")
+    blob.patch()
+    assert blob.contexts["custom"]["another"]["value"] == "value"
+
+    # Delete one context
+    blob.contexts.delete_custom_context("foo")
+    blob.patch()
+    assert "foo" not in blob.contexts["custom"] or blob.contexts["custom"]["foo"] is None
+    assert blob.contexts["custom"]["another"]["value"] == "value"
+
+    # Clear all custom contexts
+    blob.contexts.clear_custom_contexts()
+    blob.patch()
+    assert blob.contexts["custom"] is None
+
+def test_list_blobs_with_filter(shared_bucket, blobs_to_delete):
+    suffix = uuid.uuid4().hex
+    blob1_name = f"filter-test-1-{suffix}"
+    blob2_name = f"filter-test-2-{suffix}"
+
+    blob1 = shared_bucket.blob(blob1_name)
+    blob1.contexts.set_custom_context("color", "red")
+    blob1.upload_from_string(b"red-content")
+    blobs_to_delete.append(blob1)
+
+    blob2 = shared_bucket.blob(blob2_name)
+    blob2.contexts.set_custom_context("color", "blue")
+    blob2.upload_from_string(b"blue-content")
+    blobs_to_delete.append(blob2)
+
+    # Filter for red
+    # The GCS filter syntax uses 'contexts' for the field name regardless of internal SDK representation.
+    filter_expr = f'contexts.custom.color.value="red" AND name="{blob1_name}"'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 1
+    assert blobs[0].name == blob1_name
+
+    # Filter for blue
+    filter_expr = f'contexts.custom.color.value="blue" AND name="{blob2_name}"'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 1
+    assert blobs[0].name == blob2_name
+
+    # Filter for non-existent value
+    filter_expr = f'contexts.custom.color.value="green" AND name.startsWith("filter-test-")'
+    blobs = list(shared_bucket.list_blobs(filter_=filter_expr))
+
+    assert len(blobs) == 0

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

@@ -1239,6 +1239,7 @@ def test_list_blobs_w_defaults(self):
             include_folders_as_prefixes=expected_include_folders_as_prefixes,
             soft_deleted=soft_deleted,
             page_size=page_size,
+            filter_=None,
         )
 
     def test_list_blobs_w_explicit(self):
@@ -1254,6 +1255,7 @@ def test_list_blobs_w_explicit(self):
         include_folders_as_prefixes = True
         versions = True
         soft_deleted = True
+        filter_ = 'objectContexts.custom.foo.value="bar"'
         page_size = 2
         projection = "full"
         fields = "items/contentLanguage,nextPageToken"
@@ -1281,6 +1283,7 @@ def test_list_blobs_w_explicit(self):
             include_folders_as_prefixes=include_folders_as_prefixes,
             soft_deleted=soft_deleted,
             page_size=page_size,
+            filter_=filter_,
         )
 
         self.assertIs(iterator, other_client.list_blobs.return_value)
@@ -1298,6 +1301,7 @@ def test_list_blobs_w_explicit(self):
         expected_fields = fields
         expected_include_folders_as_prefixes = include_folders_as_prefixes
         expected_soft_deleted = soft_deleted
+        expected_filter = filter_
         expected_page_size = page_size
         other_client.list_blobs.assert_called_once_with(
             bucket,
@@ -1317,6 +1321,7 @@ def test_list_blobs_w_explicit(self):
             include_folders_as_prefixes=expected_include_folders_as_prefixes,
             soft_deleted=expected_soft_deleted,
             page_size=expected_page_size,
+            filter_=expected_filter,
         )
 
     def test_list_notifications_w_defaults(self):

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

@@ -2179,6 +2179,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
         include_trailing_delimiter = True
         include_folders_as_prefixes = True
         soft_deleted = False
+        filter_ = 'objectContexts.custom.foo.value="bar"'
         versions = True
         projection = "full"
         page_size = 2
@@ -2213,6 +2214,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
             match_glob=match_glob,
             include_folders_as_prefixes=include_folders_as_prefixes,
             soft_deleted=soft_deleted,
+            filter_=filter_,
         )
 
         self.assertIs(iterator, client._list_resource.return_value)
@@ -2236,6 +2238,7 @@ def test_list_blobs_w_explicit_w_user_project(self):
             "userProject": user_project,
             "includeFoldersAsPrefixes": include_folders_as_prefixes,
             "softDeleted": soft_deleted,
+            "filter": filter_,
         }
         expected_page_start = _blobs_page_start
         expected_page_size = 2

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

@@ -0,0 +1,69 @@
+import unittest
+from mock import Mock
+
+class TestObjectContexts(unittest.TestCase):
+    def _getTargetClass(self):
+        from google.cloud.storage.blob import ObjectContexts
+        return ObjectContexts
+
+    def _make_one(self, *args, **kwargs):
+        return self._getTargetClass()(*args, **kwargs)
+
+    def test_ctor(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        self.assertEqual(contexts["custom"], custom)
+        self.assertIs(contexts.blob, blob)
+
+    def test_from_api_repr(self):
+        blob = Mock()
+        resource = {"custom": {"foo": {"value": "bar"}}}
+        contexts = self._getTargetClass().from_api_repr(resource, blob)
+        self.assertEqual(contexts["custom"], resource["custom"])
+        self.assertIs(contexts.blob, blob)
+
+    def test_set_custom_context(self):
+        blob = Mock()
+        contexts = self._make_one(blob)
+        contexts.set_custom_context("foo", "bar")
+        self.assertEqual(contexts["custom"], {"foo": {"value": "bar"}})
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_delete_custom_context(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        contexts.delete_custom_context("foo")
+        self.assertIsNone(contexts["custom"]["foo"])
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_clear_custom_contexts(self):
+        blob = Mock()
+        custom = {"foo": {"value": "bar"}}
+        contexts = self._make_one(blob, custom=custom)
+        contexts.clear_custom_contexts()
+        self.assertIsNone(contexts["custom"])
+        blob._patch_property.assert_called_with("objectContexts", contexts)
+
+    def test_from_api_repr_w_timestamps(self):
+        from datetime import datetime, timezone
+        blob = Mock()
+        resource = {
+            "custom": {
+                "foo": {
+                    "value": "bar",
+                    "createTime": "2026-01-01T00:00:00.000Z",
+                    "updateTime": "2026-01-01T00:00:01.000Z",
+                }
+            }
+        }
+        contexts = self._getTargetClass().from_api_repr(resource, blob)
+        self.assertEqual(
+            contexts["custom"]["foo"]["create_time"],
+            datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
+        )
+        self.assertEqual(
+            contexts["custom"]["foo"]["update_time"],
+            datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc),
+        )