feat(storage): Add delete_source_objects optional parameter to compose API (#17163)

Add an optional `delete_source_objects` boolean parameter to the
`Blob.compose` method in `google-cloud-storage`. When this parameter is
set to `True`, it includes the `deleteSourceObjects` field in the
request body to the GCS JSON API, which triggers the deletion of source
objects upon successful composition.
Unit and system tests have been added and verified to pass.
---
*PR created automatically by Jules for task
[8742736712890332678](https://jules.google.com/task/8742736712890332678)
started by @nidhiii-27*
---------
Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: nidhiii-27 <224584462+nidhiii-27@users.noreply.github.com>

Author: nidhiii-27

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.
  """
 withcreate_trace_span(name="Storage.Blob.compose"):
 sources_len=len(sources)
@@ -3964,6 +3970,9 @@ def compose(
 "destination": self._properties.copy(),
  }
 
+ifdelete_source_objectsisnotNone:
+request["deleteSourceObjects"] =delete_source_objects
+
 ifself.user_projectisnotNone:
 query_params["userProject"] =self.user_project
 

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

@@ -1209,3 +1209,23 @@ def test_blob_download_as_bytes_single_shot_download(
 
 result_single_shot_download=blob.download_as_bytes(single_shot_download=True)
 assertresult_single_shot_download==payload
+
+
+deftest_blob_compose_delete_source_objects(shared_bucket, blobs_to_delete):
+payload_1=b"AAA\n"
+source_1=shared_bucket.blob("source-1-delete")
+source_1.upload_from_string(payload_1)
+blobs_to_delete.append(source_1)
+
+payload_2=b"BBB\n"
+source_2=shared_bucket.blob("source-2-delete")
+source_2.upload_from_string(payload_2)
+blobs_to_delete.append(source_2)
+
+destination=shared_bucket.blob("destination-delete")
+destination.compose([source_1, source_2], delete_source_objects=True)
+blobs_to_delete.append(destination)
+
+assertdestination.download_as_bytes() ==payload_1+payload_2
+assertnotsource_1.exists()
+assertnotsource_2.exists()

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

@@ -4874,6 +4874,43 @@ def test_compose_w_metageneration_match(self):
 _target_object=destination,
  )
 
+deftest_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,
+ )
+
 deftest_rewrite_w_response_wo_resource(self):
 source_name="source"
 dest_name="dest"