feat(storage): add object contexts samples

Add samples for setting, getting, listing (filter) and deleting GCS object contexts.

Region tags: storage_set_object_contexts, storage_get_object_contexts, storage_list_object_contexts, storage_delete_object_contexts.

Internal bug: b/482946642

[Generated-by: AI]

Author: sahusneha2004

storage/samples/snippets/snippets_test.py

@@ -44,6 +44,7 @@
 import storage_define_bucket_website_configuration
 import storage_delete_file
 import storage_delete_file_archived_generation
+import storage_delete_object_contexts
 import storage_disable_bucket_lifecycle_management
 import storage_disable_soft_delete
 import storage_disable_versioning
@@ -62,13 +63,15 @@
 import storage_get_bucket_labels
 import storage_get_bucket_metadata
 import storage_get_metadata
+import storage_get_object_contexts
 import storage_get_service_account
 import storage_get_soft_delete_policy
 import storage_get_soft_deleted_bucket
 import storage_list_buckets
 import storage_list_file_archived_generations
 import storage_list_files
 import storage_list_files_with_prefix
+import storage_list_object_contexts
 import storage_list_soft_deleted_buckets
 import storage_list_soft_deleted_object_versions
 import storage_list_soft_deleted_objects
@@ -85,6 +88,7 @@
 import storage_set_bucket_default_kms_key
 import storage_set_client_endpoint
 import storage_set_metadata
+import storage_set_object_contexts
 import storage_set_object_retention_policy
 import storage_set_soft_delete_policy
 import storage_trace_quickstart
@@ -248,6 +252,58 @@ def test_list_blobs(test_blob, capsys):
     assert test_blob.name in out
 
 
+def test_set_object_contexts(test_blob, capsys):
+    storage_set_object_contexts.set_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    out, _ = capsys.readouterr()
+    assert "key1 = newValue1" in out
+    assert "key3 = value3" in out
+    test_blob.reload()
+    assert "key2" not in test_blob.contexts.custom
+
+
+def test_get_object_contexts(test_blob, capsys):
+    storage_get_object_contexts.get_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    out, _ = capsys.readouterr()
+    assert f"No contexts found for {test_blob.name}" in out
+
+    storage_set_object_contexts.set_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    storage_get_object_contexts.get_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    out, _ = capsys.readouterr()
+    assert "key1 = newValue1" in out
+
+
+def test_list_object_contexts(test_blob, capsys):
+    storage_set_object_contexts.set_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    storage_list_object_contexts.list_object_contexts(
+        test_blob.bucket.name, 'contexts."key1"="newValue1"'
+    )
+    out, _ = capsys.readouterr()
+    assert test_blob.name in out
+
+
+def test_delete_object_contexts(test_blob, capsys):
+    storage_set_object_contexts.set_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    storage_delete_object_contexts.delete_object_contexts(
+        test_blob.bucket.name, test_blob.name
+    )
+    out, _ = capsys.readouterr()
+    assert f"Deleted all contexts for {test_blob.name}" in out
+    test_blob.reload()
+    assert not (test_blob.contexts and test_blob.contexts.custom)
+
+
 def test_bucket_metadata(test_bucket, capsys):
     storage_get_bucket_metadata.bucket_metadata(test_bucket.name)
     out, _ = capsys.readouterr()

storage/samples/snippets/storage_delete_object_contexts.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+
+# 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 sys
+
+# [START storage_delete_object_contexts]
+from google.cloud import storage
+
+
+def delete_object_contexts(bucket_name, blob_name):
+    """Deletes all custom contexts from an object."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+
+    # The ID of your GCS object
+    # blob_name = "your-object-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    blob = bucket.get_blob(blob_name)
+
+    # Optional: set a metageneration-match precondition to avoid potential race
+    # conditions and data corruptions. The request to patch is aborted if the
+    # object's metageneration does not match your precondition.
+    metageneration_match_precondition = blob.metageneration
+
+    blob.contexts = None
+    blob.patch(if_metageneration_match=metageneration_match_precondition)
+
+    print(f"Deleted all contexts for {blob.name}")
+
+
+# [END storage_delete_object_contexts]
+
+if __name__ == "__main__":
+    delete_object_contexts(bucket_name=sys.argv[1], blob_name=sys.argv[2])

storage/samples/snippets/storage_get_object_contexts.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+
+# 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 sys
+
+# [START storage_get_object_contexts]
+from google.cloud import storage
+
+
+def get_object_contexts(bucket_name, blob_name):
+    """Prints the custom contexts of an object."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+
+    # The ID of your GCS object
+    # blob_name = "your-object-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    blob = bucket.get_blob(blob_name)
+
+    custom_contexts = blob.contexts.custom if blob.contexts else {}
+    if not custom_contexts:
+        print(f"No contexts found for {blob.name}")
+        return
+
+    print(f"Contexts for {blob.name}:")
+    for key, payload in custom_contexts.items():
+        print(f"{key} = {payload.value}")
+
+
+# [END storage_get_object_contexts]
+
+if __name__ == "__main__":
+    get_object_contexts(bucket_name=sys.argv[1], blob_name=sys.argv[2])

storage/samples/snippets/storage_list_object_contexts.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+
+# 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 sys
+
+# [START storage_list_object_contexts]
+from google.cloud import storage
+
+
+def list_object_contexts(bucket_name, filter_string):
+    """Lists objects in a bucket filtered by object contexts."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+
+    # The filter to apply to the object contexts. Supported forms:
+    #   contexts."KEY"="VALUE"   objects with the key set to the value
+    #   contexts."KEY":*         objects that have the key
+    #   -contexts."KEY":*        objects that don't have the key
+    #   -contexts."KEY"="VALUE"  objects without the key set to the value
+    # filter_string = 'contexts."key1"="value1"'
+
+    storage_client = storage.Client()
+    blobs = storage_client.list_blobs(bucket_name, filter_=filter_string)
+
+    print(f"Objects matching filter {filter_string}:")
+    for blob in blobs:
+        print(blob.name)
+
+
+# [END storage_list_object_contexts]
+
+if __name__ == "__main__":
+    list_object_contexts(bucket_name=sys.argv[1], filter_string=sys.argv[2])

storage/samples/snippets/storage_set_object_contexts.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+
+# 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 sys
+
+# [START storage_set_object_contexts]
+from google.cloud import storage
+from google.cloud.storage.blob import ObjectCustomContextPayload
+
+
+def set_object_contexts(bucket_name, blob_name):
+    """Sets, updates and deletes custom contexts on an object."""
+    # The ID of your GCS bucket
+    # bucket_name = "your-bucket-name"
+
+    # The ID of your GCS object
+    # blob_name = "your-object-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    blob = bucket.get_blob(blob_name)
+
+    # Optional: set a metageneration-match precondition to avoid potential race
+    # conditions and data corruptions. The request to patch is aborted if the
+    # object's metageneration does not match your precondition.
+    metageneration_match_precondition = blob.metageneration
+
+    contexts = blob.contexts
+    # Add or update the values of custom contexts.
+    contexts.custom["key1"] = ObjectCustomContextPayload(value="newValue1")
+    contexts.custom["key3"] = ObjectCustomContextPayload(value="value3")
+    # Setting a context's value to None deletes that key from the object.
+    contexts.custom["key2"] = ObjectCustomContextPayload(value=None)
+    # Reassign the contexts so the change is tracked for the patch request.
+    blob.contexts = contexts
+    blob.patch(if_metageneration_match=metageneration_match_precondition)
+
+    print(f"Updated contexts for {blob.name}:")
+    for key, payload in blob.contexts.custom.items():
+        print(f"{key} = {payload.value}")
+
+
+# [END storage_set_object_contexts]
+
+if __name__ == "__main__":
+    set_object_contexts(bucket_name=sys.argv[1], blob_name=sys.argv[2])