samples(storage): add samples for soft delete objects (#1486)

Author: shubham-up-47

samples/snippets/snippets_test.py

@@ -25,8 +25,8 @@
 import requests
 
 import storage_add_bucket_label
-import storage_async_upload
 import storage_async_download
+import storage_async_upload
 import storage_batch_request
 import storage_bucket_delete_default_kms_key
 import storage_change_default_storage_class
@@ -44,6 +44,7 @@
 import storage_delete_file
 import storage_delete_file_archived_generation
 import storage_disable_bucket_lifecycle_management
+import storage_disable_soft_delete
 import storage_disable_versioning
 import storage_download_byte_range
 import storage_download_file
@@ -59,26 +60,31 @@
 import storage_get_autoclass
 import storage_get_bucket_labels
 import storage_get_bucket_metadata
-import storage_get_soft_deleted_bucket
 import storage_get_metadata
 import storage_get_service_account
+import storage_get_soft_delete_policy
+import storage_get_soft_deleted_bucket
 import storage_list_buckets
-import storage_list_soft_deleted_buckets
-import storage_restore_soft_deleted_bucket
 import storage_list_file_archived_generations
 import storage_list_files
 import storage_list_files_with_prefix
+import storage_list_soft_deleted_buckets
+import storage_list_soft_deleted_object_versions
+import storage_list_soft_deleted_objects
 import storage_make_public
 import storage_move_file
 import storage_object_get_kms_key
 import storage_remove_bucket_label
 import storage_remove_cors_configuration
 import storage_rename_file
+import storage_restore_object
+import storage_restore_soft_deleted_bucket
 import storage_set_autoclass
 import storage_set_bucket_default_kms_key
 import storage_set_client_endpoint
-import storage_set_object_retention_policy
 import storage_set_metadata
+import storage_set_object_retention_policy
+import storage_set_soft_delete_policy
 import storage_trace_quickstart
 import storage_transfer_manager_download_bucket
 import storage_transfer_manager_download_chunks_concurrently
@@ -147,6 +153,21 @@ def test_soft_deleted_bucket():
     yield bucket
 
 
+@pytest.fixture(scope="function")
+def test_soft_delete_enabled_bucket():
+    """Yields a bucket with soft-delete enabled that is deleted after the test completes."""
+    bucket = None
+    while bucket is None or bucket.exists():
+        bucket_name = f"storage-snippets-test-{uuid.uuid4()}"
+        bucket = storage.Client().bucket(bucket_name)
+    # Soft-delete retention for 7 days (minimum allowed by API)
+    bucket.soft_delete_policy.retention_duration_seconds = 7 * 24 * 60 * 60
+    # Soft-delete requires a region
+    bucket.create(location="US-CENTRAL1")
+    yield bucket
+    bucket.delete(force=True)
+
+
 @pytest.fixture(scope="function")
 def test_public_bucket():
     # The new projects don't allow to make a bucket available to public, so
@@ -230,13 +251,17 @@ def test_bucket_metadata(test_bucket, capsys):
 
 
 def test_get_soft_deleted_bucket(test_soft_deleted_bucket, capsys):
-    storage_get_soft_deleted_bucket.get_soft_deleted_bucket(test_soft_deleted_bucket.name, test_soft_deleted_bucket.generation)
+    storage_get_soft_deleted_bucket.get_soft_deleted_bucket(
+        test_soft_deleted_bucket.name, test_soft_deleted_bucket.generation
+    )
     out, _ = capsys.readouterr()
     assert test_soft_deleted_bucket.name in out
 
 
 def test_restore_soft_deleted_bucket(test_soft_deleted_bucket, capsys):
-    storage_restore_soft_deleted_bucket.restore_bucket(test_soft_deleted_bucket.name, test_soft_deleted_bucket.generation)
+    storage_restore_soft_deleted_bucket.restore_bucket(
+        test_soft_deleted_bucket.name, test_soft_deleted_bucket.generation
+    )
     out, _ = capsys.readouterr()
     assert test_soft_deleted_bucket.name in out
 
@@ -309,7 +334,9 @@ def test_async_download(test_bucket, capsys):
         blob = test_bucket.blob(source)
         blob.upload_from_string(source)
 
-    asyncio.run(storage_async_download.async_download_blobs(test_bucket.name, *source_files))
+    asyncio.run(
+        storage_async_download.async_download_blobs(test_bucket.name, *source_files)
+    )
     out, _ = capsys.readouterr()
     for x in range(object_count):
         assert f"Downloaded storage object async_sample_blob_{x}" in out
@@ -877,7 +904,10 @@ def test_object_retention_policy(test_bucket_create, capsys):
         test_bucket_create.name
     )
     out, _ = capsys.readouterr()
-    assert f"Created bucket {test_bucket_create.name} with object retention enabled setting" in out
+    assert (
+        f"Created bucket {test_bucket_create.name} with object retention enabled setting"
+        in out
+    )
 
     blob_name = "test_object_retention"
     storage_set_object_retention_policy.set_object_retention_policy(
@@ -898,7 +928,10 @@ def test_create_bucket_hierarchical_namespace(test_bucket_create, capsys):
         test_bucket_create.name
     )
     out, _ = capsys.readouterr()
-    assert f"Created bucket {test_bucket_create.name} with hierarchical namespace enabled" in out
+    assert (
+        f"Created bucket {test_bucket_create.name} with hierarchical namespace enabled"
+        in out
+    )
 
 
 def test_storage_trace_quickstart(test_bucket, capsys):
@@ -911,3 +944,96 @@ def test_storage_trace_quickstart(test_bucket, capsys):
     assert (
         f"Downloaded storage object {blob_name} from bucket {test_bucket.name}" in out
     )
+
+
+def test_storage_disable_soft_delete(test_soft_delete_enabled_bucket, capsys):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    storage_disable_soft_delete.disable_soft_delete(bucket_name)
+    out, _ = capsys.readouterr()
+    assert f"Soft-delete policy is disabled for bucket {bucket_name}" in out
+
+
+def test_storage_get_soft_delete_policy(test_soft_delete_enabled_bucket, capsys):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    storage_get_soft_delete_policy.get_soft_delete_policy(bucket_name)
+    out, _ = capsys.readouterr()
+    assert f"Soft-delete policy for {bucket_name}" in out
+    assert "Object soft-delete policy is enabled" in out
+    assert "Object retention duration: " in out
+    assert "Policy effective time: " in out
+
+    # Disable the soft-delete policy
+    test_soft_delete_enabled_bucket.soft_delete_policy.retention_duration_seconds = 0
+    test_soft_delete_enabled_bucket.patch()
+    storage_get_soft_delete_policy.get_soft_delete_policy(bucket_name)
+    out, _ = capsys.readouterr()
+    assert f"Soft-delete policy for {bucket_name}" in out
+    assert "Object soft-delete policy is disabled" in out
+
+
+def test_storage_set_soft_delete_policy(test_soft_delete_enabled_bucket, capsys):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    retention_duration_seconds = 10 * 24 * 60 * 60  # 10 days
+    storage_set_soft_delete_policy.set_soft_delete_policy(
+        bucket_name, retention_duration_seconds
+    )
+    out, _ = capsys.readouterr()
+    assert (
+        f"Soft delete policy for bucket {bucket_name} was set to {retention_duration_seconds} seconds retention period"
+        in out
+    )
+
+
+def test_storage_list_soft_deleted_objects(test_soft_delete_enabled_bucket, capsys):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    blob_name = f"test_object_{uuid.uuid4().hex}.txt"
+    blob_content = "This object will be soft-deleted for listing."
+    blob = test_soft_delete_enabled_bucket.blob(blob_name)
+    blob.upload_from_string(blob_content)
+    blob_generation = blob.generation
+
+    blob.delete()  # Soft-delete the object
+    storage_list_soft_deleted_objects.list_soft_deleted_objects(bucket_name)
+    out, _ = capsys.readouterr()
+    assert f"Name: {blob_name}, Generation: {blob_generation}" in out
+
+
+def test_storage_list_soft_deleted_object_versions(
+    test_soft_delete_enabled_bucket, capsys
+):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    blob_name = f"test_object_{uuid.uuid4().hex}.txt"
+    blob_content = "This object will be soft-deleted for version listing."
+    blob = test_soft_delete_enabled_bucket.blob(blob_name)
+    blob.upload_from_string(blob_content)
+    blob_generation = blob.generation
+
+    blob.delete()  # Soft-delete the object
+    storage_list_soft_deleted_object_versions.list_soft_deleted_object_versions(
+        bucket_name, blob_name
+    )
+    out, _ = capsys.readouterr()
+    assert f"Version ID: {blob_generation}" in out
+
+
+def test_storage_restore_soft_deleted_object(test_soft_delete_enabled_bucket, capsys):
+    bucket_name = test_soft_delete_enabled_bucket.name
+    blob_name = f"test-restore-sd-obj-{uuid.uuid4().hex}.txt"
+    blob_content = "This object will be soft-deleted and restored."
+    blob = test_soft_delete_enabled_bucket.blob(blob_name)
+    blob.upload_from_string(blob_content)
+    blob_generation = blob.generation
+
+    blob.delete()  # Soft-delete the object
+    storage_restore_object.restore_soft_deleted_object(
+        bucket_name, blob_name, blob_generation
+    )
+    out, _ = capsys.readouterr()
+    assert (
+        f"Soft-deleted object {blob_name} is restored in the bucket {bucket_name}"
+        in out
+    )
+
+    # Verify the restoration
+    blob = test_soft_delete_enabled_bucket.get_blob(blob_name)
+    assert blob is not None

samples/snippets/storage_disable_soft_delete.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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_disable_soft_delete]
+from google.cloud import storage
+
+
+def disable_soft_delete(bucket_name):
+    """Disable soft-delete policy for the bucket."""
+    # bucket_name = "your-bucket-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    # Setting the retention duration to 0 disables soft-delete.
+    bucket.soft_delete_policy.retention_duration_seconds = 0
+    bucket.patch()
+
+    print(f"Soft-delete policy is disabled for bucket {bucket_name}")
+
+
+# [END storage_disable_soft_delete]
+
+if __name__ == "__main__":
+    disable_soft_delete(bucket_name=sys.argv[1])

samples/snippets/storage_get_soft_delete_policy.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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_soft_delete_policy]
+from google.cloud import storage
+
+
+def get_soft_delete_policy(bucket_name):
+    """Gets the soft-delete policy of the bucket"""
+    # bucket_name = "your-bucket-name"
+
+    storage_client = storage.Client()
+    bucket = storage_client.get_bucket(bucket_name)
+
+    print(f"Soft-delete policy for {bucket_name}")
+    if (
+        bucket.soft_delete_policy
+        and bucket.soft_delete_policy.retention_duration_seconds
+    ):
+        print("Object soft-delete policy is enabled")
+        print(
+            f"Object retention duration: {bucket.soft_delete_policy.retention_duration_seconds} seconds"
+        )
+        print(f"Policy effective time: {bucket.soft_delete_policy.effective_time}")
+    else:
+        print("Object soft-delete policy is disabled")
+
+
+# [END storage_get_soft_delete_policy]
+
+if __name__ == "__main__":
+    get_soft_delete_policy(bucket_name=sys.argv[1])

samples/snippets/storage_list_soft_deleted_object_versions.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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_soft_deleted_object_versions]
+from google.cloud import storage
+
+
+def list_soft_deleted_object_versions(bucket_name, blob_name):
+    """Lists all versions of a soft-deleted object in the bucket."""
+    # bucket_name = "your-bucket-name"
+    # blob_name = "your-object-name"
+
+    storage_client = storage.Client()
+    blobs = storage_client.list_blobs(bucket_name, prefix=blob_name, soft_deleted=True)
+
+    # Note: The call returns a response only when the iterator is consumed.
+    for blob in blobs:
+        print(
+            f"Version ID: {blob.generation}, Soft Delete Time: {blob.soft_delete_time}"
+        )
+
+
+# [END storage_list_soft_deleted_object_versions]
+
+if __name__ == "__main__":
+    list_soft_deleted_object_versions(bucket_name=sys.argv[1], blob_name=sys.argv[2])

samples/snippets/storage_list_soft_deleted_objects.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+# Copyright 2025 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_soft_deleted_objects]
+from google.cloud import storage
+
+
+def list_soft_deleted_objects(bucket_name):
+    """Lists all soft-deleted objects in the bucket."""
+    # bucket_name = "your-bucket-name"
+
+    storage_client = storage.Client()
+    blobs = storage_client.list_blobs(bucket_name, soft_deleted=True)
+
+    # Note: The call returns a response only when the iterator is consumed.
+    for blob in blobs:
+        print(
+            f"Name: {blob.name}, Generation: {blob.generation}, Soft Delete Time: {blob.soft_delete_time}"
+        )
+
+
+# [END storage_list_soft_deleted_objects]
+
+if __name__ == "__main__":
+    list_soft_deleted_objects(bucket_name=sys.argv[1])