feat(storage): support delete_source_objects in compose API

Implements the delete_source_objects parameter in Blob.compose() to allow
automatic deletion of source objects upon successful composition.

Reference PR: googleapis/google-cloud-java#12873

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

Author: google-labs-jules[bot]

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

@@ -3849,6 +3849,7 @@ def compose(
         if_metageneration_match=None,
         if_source_generation_match=None,
         retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+        delete_source_objects=None,
     ):
         """Concatenate source blobs into this one.
 
@@ -3908,6 +3909,11 @@ def compose(
             Change the value to ``DEFAULT_RETRY`` or another `google.api_core.retry.Retry` object
             to enable retries regardless of generation precondition setting.
             See [Configuring Retries](https://cloud.google.com/python/docs/reference/storage/latest/retry_timeout).
+
+        :type delete_source_objects: bool
+        :param delete_source_objects:
+            (Optional) If True, the source objects will be deleted after a
+            successful composition.
         """
         with create_trace_span(name="Storage.Blob.compose"):
             sources_len = len(sources)
@@ -3964,6 +3970,9 @@ def compose(
                 "destination": self._properties.copy(),
             }
 
+            if delete_source_objects is not None:
+                request["deleteSourceObjects"] = delete_source_objects
+
             if self.user_project is not None:
                 query_params["userProject"] = self.user_project
 

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

@@ -4480,6 +4480,43 @@ def test_compose_wo_content_type_set(self):
             _target_object=destination,
         )
 
+    def test_compose_w_delete_source_objects(self):
+        source_1_name = "source-1"
+        source_2_name = "source-2"
+        destination_name = "destination"
+        delete_source_objects = True
+        api_response = {}
+        client = mock.Mock(spec=["_post_resource"])
+        client._post_resource.return_value = api_response
+        bucket = _Bucket(client=client)
+        source_1 = self._make_one(source_1_name, bucket=bucket)
+        source_2 = self._make_one(source_2_name, bucket=bucket)
+        destination = self._make_one(destination_name, bucket=bucket)
+
+        destination.compose(
+            sources=[source_1, source_2],
+            delete_source_objects=delete_source_objects,
+        )
+
+        expected_path = f"/b/name/o/{destination_name}/compose"
+        expected_data = {
+            "sourceObjects": [
+                {"name": source_1.name, "generation": source_1.generation},
+                {"name": source_2.name, "generation": source_2.generation},
+            ],
+            "destination": {},
+            "deleteSourceObjects": delete_source_objects,
+        }
+        expected_query_params = {}
+        client._post_resource.assert_called_once_with(
+            expected_path,
+            expected_data,
+            query_params=expected_query_params,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            _target_object=destination,
+        )
+
     def test_compose_minimal_w_user_project_w_timeout(self):
         source_1_name = "source-1"
         source_2_name = "source-2"