diff --git a/packages/google-cloud-storage/google/cloud/storage/blob.py b/packages/google-cloud-storage/google/cloud/storage/blob.py
index c6fbcf4c12b7..32a648a938eb 100644
--- a/packages/google-cloud-storage/google/cloud/storage/blob.py
+++ b/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
 
diff --git a/packages/google-cloud-storage/tests/system/test_blob.py b/packages/google-cloud-storage/tests/system/test_blob.py
index 60a2fa2568b2..c3cb1204fbd6 100644
--- a/packages/google-cloud-storage/tests/system/test_blob.py
+++ b/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, 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)
+
+    assert destination.download_as_bytes() == payload_1 + payload_2
+    assert not source_1.exists()
+    assert not source_2.exists()
+
+
 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")
diff --git a/packages/google-cloud-storage/tests/unit/test_blob.py b/packages/google-cloud-storage/tests/unit/test_blob.py
index a218f011dd17..942f594669b6 100644
--- a/packages/google-cloud-storage/tests/unit/test_blob.py
+++ b/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"