feat(storage): support delete_source_objects in compose API (Address PR feedback)

- Removed unnecessary fixture usage in system test.
- Ensured destination metadata is not empty in unit test by setting content_type.
- verified end-to-end functionality in system test.

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/system/test_blob.py

@@ -853,6 +853,26 @@ def test_blob_compose_new_blob(shared_bucket, blobs_to_delete):
     assert destination.download_as_bytes() == payload_1 + payload_2
 
 
+def test_blob_compose_delete_source_objects(shared_bucket):
+    payload_1 = b"AAA\n"
+    source_1 = shared_bucket.blob("source-1-delete")
+    source_1.upload_from_string(payload_1)
+
+    payload_2 = b"BBB\n"
+    source_2 = shared_bucket.blob("source-2-delete")
+    source_2.upload_from_string(payload_2)
+
+    destination = shared_bucket.blob("destination-delete")
+    destination.compose([source_1, source_2], delete_source_objects=True)
+
+    try:
+        assert destination.download_as_bytes() == payload_1 + payload_2
+        assert not source_1.exists()
+        assert not source_2.exists()
+    finally:
+        destination.delete()
+
+
 def test_blob_compose_new_blob_wo_content_type(shared_bucket, blobs_to_delete):
     payload_1 = b"AAA\n"
     source_1 = shared_bucket.blob("source-1")

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

@@ -4480,6 +4480,45 @@ 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"
+        content_type = "text/plain"
+        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.content_type = content_type
+
+        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": {"contentType": content_type},
+            "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"