Updating S3 Transfer Manager for deletion

Adding response to delete handler

Author: Rakshil Modi

awscli/s3transfer/constants.py

@@ -36,3 +36,4 @@
 
 USER_AGENT = f's3transfer/{s3transfer.__version__}'
 PROCESS_USER_AGENT = f'{USER_AGENT} processpool'
+MAX_BATCH_SIZE = 1000

awscli/s3transfer/delete.py

@@ -10,6 +10,7 @@
 # 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.
+
 from s3transfer.tasks import SubmissionTask, Task
 
 
@@ -69,3 +70,78 @@ def _main(self, client, bucket, key, extra_args):
 
         """
         client.delete_object(Bucket=bucket, Key=key, **extra_args)
+
+
+class BatchDeleteSubmissionTask(SubmissionTask):
+    """Task for submitting tasks to execute batch deletion of S3 objects"""
+
+    def _submit(self, client, request_executor, transfer_future, **kwargs):
+        """
+        :param client: The client associated with the transfer manager
+
+        :type config: s3transfer.manager.TransferConfig
+        :param config: The transfer config associated with the transfer
+            manager
+
+        :type osutil: s3transfer.utils.OSUtil
+        :param osutil: The os utility associated to the transfer manager
+
+        :type request_executor: s3transfer.futures.BoundedExecutor
+        :param request_executor: The request executor associated with the
+            transfer manager
+
+        :type transfer_future: s3transfer.futures.TransferFuture
+        :param transfer_future: The transfer future associated with the
+            transfer request that tasks are being submitted for
+        """
+        call_args = transfer_future.meta.call_args
+
+        # Extract the objects to delete from the call_args
+        bucket = call_args.bucket
+        objects = call_args.objects
+        extra_args = call_args.extra_args or {}
+
+        self._transfer_coordinator.submit(
+            request_executor,
+            BatchDeleteObjectsTask(
+                transfer_coordinator=self._transfer_coordinator,
+                main_kwargs={
+                    'client': client,
+                    'bucket': bucket,
+                    'objects': objects,
+                    'extra_args': extra_args,
+                },
+                is_final=True,
+            ),
+        )
+
+
+class BatchDeleteObjectsTask(Task):
+    def _main(self, client, bucket, objects, extra_args):
+        """
+        Delete multiple objects in a single API call
+
+        :param client: The S3 client to use when calling DeleteObjects
+
+        :type bucket: str
+        :param bucket: The name of the bucket.
+
+        :type objects: list
+        :param objects: The list of objects to delete.
+
+        :type extra_args: dict
+        :param extra_args: Extra arguments to pass to the DeleteObjects call.
+        """
+        # Create a copy of extra_args to avoid modifying the original
+        extra_args_copy = extra_args.copy()
+
+        # Create the request body
+        delete_request = {
+            'Objects': objects,
+            'Quiet': False,
+        }
+
+        response = client.delete_objects(
+            Bucket=bucket, Delete=delete_request, **extra_args_copy
+        )
+        return response

awscli/s3transfer/manager.py

@@ -24,7 +24,7 @@
     MB,
 )
 from s3transfer.copies import CopySubmissionTask
-from s3transfer.delete import DeleteSubmissionTask
+from s3transfer.delete import BatchDeleteSubmissionTask, DeleteSubmissionTask
 from s3transfer.download import DownloadSubmissionTask
 from s3transfer.exceptions import CancelledError, FatalError
 from s3transfer.futures import (
@@ -665,6 +665,26 @@ def _shutdown(self, cancel, cancel_msg, exc_type=CancelledError):
             self._request_executor.shutdown()
             self._io_executor.shutdown()
 
+    def batch_delete(self, bucket, objects, extra_args=None, subscribers=None):
+        """Delete multiple S3 objects in batches"""
+        if extra_args is None:
+            extra_args = {}
+        if subscribers is None:
+            subscribers = []
+        # Version ID is not included in input shape.
+        allowed_delete_objects_args = [
+            arg for arg in self.ALLOWED_DELETE_ARGS if arg != 'VersionId'
+        ]
+        self._validate_all_known_args(extra_args, allowed_delete_objects_args)
+        self._validate_if_bucket_supported(bucket)
+        call_args = CallArgs(
+            bucket=bucket,
+            objects=objects,
+            extra_args=extra_args,
+            subscribers=subscribers,
+        )
+        return self._submit_transfer(call_args, BatchDeleteSubmissionTask)
+
 
 class TransferCoordinatorController:
     def __init__(self):

tests/functional/s3transfer/test_delete.py

@@ -74,3 +74,76 @@ def test_raise_exception_on_s3_object_lambda_resource(self):
         )
         with self.assertRaisesRegex(ValueError, 'methods do not support'):
             self.manager.delete(s3_object_lambda_arn, self.key)
+
+
+class TestBatchDelete(BaseGeneralInterfaceTest):
+    __test__ = True
+
+    def setUp(self):
+        super().setUp()
+        self.bucket = 'mybucket'
+        self.key = 'mykey'
+        self.objects = [
+            {'Key': self.key, 'VersionId': 'version1'},
+            {'Key': self.key, 'VersionId': 'deletemarker1'},
+        ]
+        self.manager = TransferManager(self.client)
+
+    @property
+    def method(self):
+        """The transfer manager method to invoke i.e. upload()"""
+        return self.manager.batch_delete
+
+    def create_call_kwargs(self):
+        """The kwargs to be passed to the transfer manager method"""
+        return {
+            'bucket': self.bucket,
+            'objects': self.objects,
+        }
+
+    def create_invalid_extra_args(self):
+        return {
+            'BadKwargs': True,
+        }
+
+    def create_stubbed_responses(self):
+        """A list of stubbed responses that will cause the request to succeed
+
+        The elements of this list is a dictionary that will be used as key
+        word arguments to botocore.Stubber.add_response(). For example::
+
+            [{'method': 'delete_objects', 'service_response': {}}]
+        """
+        return [
+            {
+                'method': 'delete_objects',
+                'service_response': {},
+                'expected_params': {
+                    'Bucket': self.bucket,
+                    'Delete': {'Objects': self.objects, 'Quiet': False},
+                },
+            }
+        ]
+
+    def create_expected_progress_callback_info(self):
+        return []
+
+    def test_known_allowed_args_in_input_shape(self):
+        op_model = self.client.meta.service_model.operation_model(
+            'DeleteObjects'
+        )
+        allowed_delete_objects_args = [
+            arg
+            for arg in self.manager.ALLOWED_DELETE_ARGS
+            if arg != 'VersionId'
+        ]
+        for allowed_arg in allowed_delete_objects_args:
+            self.assertIn(allowed_arg, op_model.input_shape.members)
+
+    def test_raise_exception_on_s3_object_lambda_resource(self):
+        s3_object_lambda_arn = (
+            'arn:aws:s3-object-lambda:us-west-2:123456789012:'
+            'accesspoint:my-accesspoint'
+        )
+        with self.assertRaisesRegex(ValueError, 'methods do not support'):
+            self.manager.batch_delete(s3_object_lambda_arn, self.key)

tests/integration/s3transfer/test_delete.py

@@ -26,3 +26,22 @@ def test_can_delete_object(self):
         future.result()
 
         self.assertTrue(self.object_not_exists(key_name))
+
+
+class TestBatchDeleteObject(BaseTransferManagerIntegTest):
+    def test_can_delete_versioned_objects(self):
+        key_name = 'mykey'
+        self.client.put_object(
+            Bucket=self.bucket_name, Key=key_name, Body=b'hello world'
+        )
+        response = self.client.list_object_versions(Bucket=self.bucket_name)
+        version_id = response['Versions'][0]['VersionId']
+        objects = [{'Key': key_name, 'VersionId': version_id}]
+        transfer_manager = self.create_transfer_manager()
+        future = transfer_manager.batch_delete(
+            bucket=self.bucket_name, objects=objects
+        )
+
+        future.result()
+
+        self.assertTrue(self.object_not_exists(key_name))