Add Object Contexts and List Filter Support

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

@@ -19,6 +19,7 @@
     "content_type": "content_type",
     "metadata": "metadata",
     "kms_key_name": "kms_key",
+    "contexts": "contexts",
 }
 
 

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

@@ -98,6 +98,7 @@
     "contentEncoding",
     "contentLanguage",
     _CONTENT_TYPE_FIELD,
+    "contexts",
     "crc32c",
     "customTime",
     "md5Hash",
@@ -5008,6 +5009,16 @@ def retention(self):
         info = self._properties.get("retention", {})
         return Retention.from_api_repr(info, self)
 
+    @property
+    def contexts(self):
+        """Retrieve the contexts configuration for this object.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: an instance for managing the object's contexts configuration.
+        """
+        info = self._properties.get("contexts", {})
+        return ObjectContexts.from_api_repr(info, self)
+
     @property
     def soft_delete_time(self):
         """If this object has been soft-deleted, returns the time at which it became soft-deleted.
@@ -5300,3 +5311,62 @@ def retention_expiration_time(self):
         retention_expiration_time = self.get("retentionExpirationTime")
         if retention_expiration_time is not None:
             return _rfc3339_nanos_to_datetime(retention_expiration_time)
+
+
+class ObjectContexts(dict):
+    """Map an object's contexts configuration.
+
+    :type blob: :class:`Blob`
+    :param blob: blob for which this contexts configuration applies to.
+    """
+
+    def __init__(self, blob):
+        super(ObjectContexts, self).__init__({"custom": {}})
+        self._blob = blob
+
+    @classmethod
+    def from_api_repr(cls, resource, blob):
+        """Factory:  construct instance from resource.
+
+        :type blob: :class:`Blob`
+        :param blob: Blob for which this contexts configuration applies to.
+
+        :type resource: dict
+        :param resource: mapping as returned from API call.
+
+        :rtype: :class:`ObjectContexts`
+        :returns: ObjectContexts configuration created from resource.
+        """
+        instance = cls(blob)
+        instance.update(resource)
+        return instance
+
+    @property
+    def blob(self):
+        """Blob for which this contexts configuration applies to.
+
+        :rtype: :class:`Blob`
+        :returns: the instance's blob.
+        """
+        return self._blob
+
+    def set_custom_context(self, key, value):
+        """Sets a custom context key-value pair.
+
+        :type key: str
+        :param key: The key of the custom context.
+
+        :type value: str
+        :param value: The value of the custom context.
+        """
+        self.setdefault("custom", {})[key] = {"value": value}
+        self.blob._patch_property("contexts", self)
+
+    def delete_custom_context(self, key):
+        """Deletes a custom context key-value pair.
+
+        :type key: str
+        :param key: The key of the custom context to delete.
+        """
+        self.setdefault("custom", {})[key] = {"delete": True}
+        self.blob._patch_property("contexts", self)

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

@@ -1423,6 +1423,7 @@ 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 +1522,11 @@ 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 used to filter results.
+            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 +1551,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,7 @@ 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 +1401,10 @@ 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 used to filter results.
+                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
@@ -1443,6 +1448,9 @@ def list_blobs(
             if soft_deleted is not None:
                 extra_params["softDeleted"] = soft_deleted
 
+            if filter is not None:
+                extra_params["filter"] = filter
+
             if bucket.user_project is not None:
                 extra_params["userProject"] = bucket.user_project
 

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

@@ -0,0 +1,46 @@
+# Copyright 2026 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
+#
+#     http://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.
+
+import unittest
+from unittest import mock
+
+
+class Test_blob_to_proto(unittest.TestCase):
+    def _call_fut(self, blob):
+        from google.cloud.storage._grpc_conversions import blob_to_proto
+
+        return blob_to_proto(blob)
+
+    def test_w_contexts(self):
+        from google.cloud.storage.blob import Blob
+        from google.cloud.storage.bucket import Bucket
+
+        blob_name = "blob-name"
+        bucket_name = "bucket-name"
+        bucket = mock.Mock(spec=Bucket)
+        bucket.name = bucket_name
+        contexts = {"custom": {"foo": {"value": "bar"}}}
+        blob = mock.Mock(spec=Blob)
+        blob.name = blob_name
+        blob.bucket = bucket
+        blob.content_type = None
+        blob.metadata = None
+        blob.kms_key_name = None
+        blob.contexts = contexts
+
+        proto = self._call_fut(blob)
+
+        self.assertEqual(proto.name, blob_name)
+        self.assertEqual(proto.bucket, f"projects/_/buckets/{bucket_name}")
+        self.assertEqual(proto.contexts.custom["foo"].value, "bar")

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

@@ -6205,6 +6205,51 @@ def test_object_lock_retention_configuration_setter(self):
         self.assertIsNone(blob.retention.retain_until_time)
         self.assertIn("retention", blob._changes)
 
+    def test_contexts_defaults(self):
+        from google.cloud.storage.blob import ObjectContexts
+
+        BLOB_NAME = "blob-name"
+        BUCKET = object()
+        blob = self._make_one(BLOB_NAME, bucket=BUCKET)
+
+        contexts = blob.contexts
+
+        self.assertIsInstance(contexts, ObjectContexts)
+        self.assertIs(contexts.blob, blob)
+        self.assertEqual(contexts, {"custom": {}})
+
+    def test_contexts_w_entry(self):
+        from google.cloud.storage.blob import ObjectContexts
+
+        properties = {"contexts": {"custom": {"foo": {"value": "bar"}}}}
+        BLOB_NAME = "blob-name"
+        BUCKET = object()
+        blob = self._make_one(BLOB_NAME, bucket=BUCKET, properties=properties)
+
+        contexts = blob.contexts
+
+        self.assertIsInstance(contexts, ObjectContexts)
+        self.assertIs(contexts.blob, blob)
+        self.assertEqual(contexts, properties["contexts"])
+
+    def test_contexts_set_custom_context(self):
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+
+        blob.contexts.set_custom_context("foo", "bar")
+        self.assertEqual(blob.contexts["custom"]["foo"], {"value": "bar"})
+        self.assertIn("contexts", blob._changes)
+
+    def test_contexts_delete_custom_context(self):
+        BLOB_NAME = "blob-name"
+        bucket = _Bucket()
+        blob = self._make_one(BLOB_NAME, bucket=bucket)
+
+        blob.contexts.delete_custom_context("foo")
+        self.assertEqual(blob.contexts["custom"]["foo"], {"delete": True})
+        self.assertIn("contexts", blob._changes)
+
 
 class Test__quote(unittest.TestCase):
     @staticmethod

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):
@@ -1317,6 +1318,37 @@ 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=None,
+        )
+
+    def test_list_blobs_w_filter(self):
+        name = "name"
+        client = self._make_client()
+        client.list_blobs = mock.Mock(spec=[])
+        bucket = self._make_one(client=client, name=name)
+        filter_expr = 'custom.foo = "bar"'
+
+        bucket.list_blobs(filter=filter_expr)
+
+        client.list_blobs.assert_called_once_with(
+            bucket,
+            max_results=None,
+            page_token=None,
+            prefix=None,
+            delimiter=None,
+            start_offset=None,
+            end_offset=None,
+            include_trailing_delimiter=None,
+            versions=None,
+            projection="noAcl",
+            fields=None,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
+            match_glob=None,
+            include_folders_as_prefixes=None,
+            soft_deleted=None,
+            page_size=None,
+            filter=filter_expr,
         )
 
     def test_list_notifications_w_defaults(self):

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

@@ -2251,6 +2251,40 @@ def test_list_blobs_w_explicit_w_user_project(self):
             retry=retry,
         )
 
+    def test_list_blobs_w_filter(self):
+        from google.cloud.storage.bucket import _blobs_page_start, _item_to_blob
+
+        project = "PROJECT"
+        bucket_name = "name"
+        filter_expr = 'custom.foo = "bar"'
+        credentials = _make_credentials()
+        client = self._make_one(project=project, credentials=credentials)
+        client._list_resource = mock.Mock(spec=[])
+        client._bucket_arg_to_bucket = mock.Mock(spec=[])
+        bucket = client._bucket_arg_to_bucket.return_value = mock.Mock(
+            spec=["path", "user_project"],
+        )
+        bucket.path = f"/b/{bucket_name}"
+        bucket.user_project = None
+
+        client.list_blobs(bucket_name, filter=filter_expr)
+
+        expected_extra_params = {
+            "projection": "noAcl",
+            "filter": filter_expr,
+        }
+        client._list_resource.assert_called_once_with(
+            f"/b/{bucket_name}/o",
+            _item_to_blob,
+            page_token=None,
+            max_results=None,
+            extra_params=expected_extra_params,
+            page_start=_blobs_page_start,
+            page_size=None,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
+        )
+
     def test_list_buckets_wo_project(self):
         from google.cloud.exceptions import BadRequest