class BandwidthLimiter:
    def __init__(self, leaky_bucket, time_utils=None):
        """Limits bandwidth for shared S3 transfers

        :type leaky_bucket: LeakyBucket
        :param leaky_bucket: The leaky bucket to use limit bandwidth

        :type time_utils: TimeUtils
        :param time_utils: Time utility to use for interacting with time.
        """
        self._leaky_bucket = leaky_bucket
        self._time_utils = time_utils
        if time_utils is None:
            self._time_utils = TimeUtils()

    def get_bandwith_limited_stream(
        self, fileobj, transfer_coordinator, enabled=True
    ):
        """Wraps a fileobj in a bandwidth limited stream wrapper

        :type fileobj: file-like obj
        :param fileobj: The file-like obj to wrap

        :type transfer_coordinator: s3transfer.futures.TransferCoordinator
        param transfer_coordinator: The coordinator for the general transfer
            that the wrapped stream is a part of

        :type enabled: boolean
        :param enabled: Whether bandwidth limiting should be enabled to start
        """
        stream = BandwidthLimitedStream(
            fileobj, self._leaky_bucket, transfer_coordinator, self._time_utils
        )
        if not enabled:
            stream.disable_bandwidth_limiting()
        return stream class LeakyBucket:
    def __init__(
        self,
        max_rate,
        time_utils=None,
        rate_tracker=None,
        consumption_scheduler=None,
    ):
        """A leaky bucket abstraction to limit bandwidth consumption

        :type rate: int
        :type rate: The maximum rate to allow. This rate is in terms of
            bytes per second.

        :type time_utils: TimeUtils
        :param time_utils: The time utility to use for interacting with time

        :type rate_tracker: BandwidthRateTracker
        :param rate_tracker: Tracks bandwidth consumption

        :type consumption_scheduler: ConsumptionScheduler
        :param consumption_scheduler: Schedules consumption retries when
            necessary
        """
        self._max_rate = float(max_rate)
        self._time_utils = time_utils
        if time_utils is None:
            self._time_utils = TimeUtils()
        self._lock = threading.Lock()
        self._rate_tracker = rate_tracker
        if rate_tracker is None:
            self._rate_tracker = BandwidthRateTracker()
        self._consumption_scheduler = consumption_scheduler
        if consumption_scheduler is None:
            self._consumption_scheduler = ConsumptionScheduler()

    def consume(self, amt, request_token):
        """Consume an a requested amount

        :type amt: int
        :param amt: The amount of bytes to request to consume

        :type request_token: RequestToken
        :param request_token: The token associated to the consumption
            request that is used to identify the request. So if a
            RequestExceededException is raised the token should be used
            in subsequent retry consume() request.

        :raises RequestExceededException: If the consumption amount would
            exceed the maximum allocated bandwidth

        :rtype: int
        :returns: The amount consumed
        """
        with self._lock:
            time_now = self._time_utils.time()
            if self._consumption_scheduler.is_scheduled(request_token):
                return self._release_requested_amt_for_scheduled_request(
                    amt, request_token, time_now
                )
            elif self._projected_to_exceed_max_rate(amt, time_now):
                self._raise_request_exceeded_exception(
                    amt, request_token, time_now
                )
            else:
                return self._release_requested_amt(amt, time_now)

    def _projected_to_exceed_max_rate(self, amt, time_now):
        projected_rate = self._rate_tracker.get_projected_rate(amt, time_now)
        return projected_rate > self._max_rate

    def _release_requested_amt_for_scheduled_request(
        self, amt, request_token, time_now
    ):
        self._consumption_scheduler.process_scheduled_consumption(
            request_token
        )
        return self._release_requested_amt(amt, time_now)

    def _raise_request_exceeded_exception(self, amt, request_token, time_now):
        allocated_time = amt / float(self._max_rate)
        retry_time = self._consumption_scheduler.schedule_consumption(
            amt, request_token, allocated_time
        )
        raise RequestExceededException(
            requested_amt=amt, retry_time=retry_time
        )

    def _release_requested_amt(self, amt, time_now):
        self._rate_tracker.record_consumption_rate(amt, time_now)
        return amt ALLOWED_DOWNLOAD_ARGS = [
    'ChecksumMode',
    'VersionId',
    'SSECustomerAlgorithm',
    'SSECustomerKey',
    'SSECustomerKeyMD5',
    'RequestPayer',
    'ExpectedBucketOwner',
] KB = 1024 MB = KB * KB class CopySubmissionTask(SubmissionTask):
    """Task for submitting tasks to execute a copy"""

    EXTRA_ARGS_TO_HEAD_ARGS_MAPPING = {
        'CopySourceIfMatch': 'IfMatch',
        'CopySourceIfModifiedSince': 'IfModifiedSince',
        'CopySourceIfNoneMatch': 'IfNoneMatch',
        'CopySourceIfUnmodifiedSince': 'IfUnmodifiedSince',
        'CopySourceSSECustomerKey': 'SSECustomerKey',
        'CopySourceSSECustomerAlgorithm': 'SSECustomerAlgorithm',
        'CopySourceSSECustomerKeyMD5': 'SSECustomerKeyMD5',
        'RequestPayer': 'RequestPayer',
        'ExpectedBucketOwner': 'ExpectedBucketOwner',
    }

    UPLOAD_PART_COPY_ARGS = [
        'CopySourceIfMatch',
        'CopySourceIfModifiedSince',
        'CopySourceIfNoneMatch',
        'CopySourceIfUnmodifiedSince',
        'CopySourceSSECustomerKey',
        'CopySourceSSECustomerAlgorithm',
        'CopySourceSSECustomerKeyMD5',
        'SSECustomerKey',
        'SSECustomerAlgorithm',
        'SSECustomerKeyMD5',
        'RequestPayer',
        'ExpectedBucketOwner',
    ]

    CREATE_MULTIPART_ARGS_BLACKLIST = [
        'CopySourceIfMatch',
        'CopySourceIfModifiedSince',
        'CopySourceIfNoneMatch',
        'CopySourceIfUnmodifiedSince',
        'CopySourceSSECustomerKey',
        'CopySourceSSECustomerAlgorithm',
        'CopySourceSSECustomerKeyMD5',
        'MetadataDirective',
        'TaggingDirective',
    ]

    COMPLETE_MULTIPART_ARGS = ['RequestPayer', 'ExpectedBucketOwner']

    def _submit(
        self, client, config, osutil, request_executor, transfer_future
    ):
        """
        :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
        """
        # Determine the size if it was not provided
        if transfer_future.meta.size is None:
            # If a size was not provided figure out the size for the
            # user. Note that we will only use the client provided to
            # the TransferManager. If the object is outside of the region
            # of the client, they may have to provide the file size themselves
            # with a completely new client.
            call_args = transfer_future.meta.call_args
            head_object_request = (
                self._get_head_object_request_from_copy_source(
                    call_args.copy_source
                )
            )
            extra_args = call_args.extra_args

            # Map any values that may be used in the head object that is
            # used in the copy object
            for param, value in extra_args.items():
                if param in self.EXTRA_ARGS_TO_HEAD_ARGS_MAPPING:
                    head_object_request[
                        self.EXTRA_ARGS_TO_HEAD_ARGS_MAPPING[param]
                    ] = value

            response = call_args.source_client.head_object(
                **head_object_request
            )
            transfer_future.meta.provide_transfer_size(
                response['ContentLength']
            )

        # If it is greater than threshold do a multipart copy, otherwise
        # do a regular copy object.
        if transfer_future.meta.size < config.multipart_threshold:
            self._submit_copy_request(
                client, config, osutil, request_executor, transfer_future
            )
        else:
            self._submit_multipart_request(
                client, config, osutil, request_executor, transfer_future
            )

    def _submit_copy_request(
        self, client, config, osutil, request_executor, transfer_future
    ):
        call_args = transfer_future.meta.call_args

        # Get the needed progress callbacks for the task
        progress_callbacks = get_callbacks(transfer_future, 'progress')

        # Submit the request of a single copy.
        self._transfer_coordinator.submit(
            request_executor,
            CopyObjectTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'copy_source': call_args.copy_source,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': call_args.extra_args,
                    'callbacks': progress_callbacks,
                    'size': transfer_future.meta.size,
                },
                is_final=True,
            ),
        )

    def _submit_multipart_request(
        self, client, config, osutil, request_executor, transfer_future
    ):
        call_args = transfer_future.meta.call_args

        # Submit the request to create a multipart upload and make sure it
        # does not include any of the arguments used for copy part.
        create_multipart_extra_args = {}
        for param, val in call_args.extra_args.items():
            if param not in self.CREATE_MULTIPART_ARGS_BLACKLIST:
                create_multipart_extra_args[param] = val

        create_multipart_future = self._transfer_coordinator.submit(
            request_executor,
            CreateMultipartUploadTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': create_multipart_extra_args,
                },
            ),
        )

        # Determine how many parts are needed based on filesize and
        # desired chunksize.
        part_size = config.multipart_chunksize
        adjuster = ChunksizeAdjuster()
        part_size = adjuster.adjust_chunksize(
            part_size, transfer_future.meta.size
        )
        num_parts = int(
            math.ceil(transfer_future.meta.size / float(part_size))
        )

        # Submit requests to upload the parts of the file.
        part_futures = []
        progress_callbacks = get_callbacks(transfer_future, 'progress')

        for part_number in range(1, num_parts + 1):
            extra_part_args = self._extra_upload_part_args(
                call_args.extra_args
            )
            # The part number for upload part starts at 1 while the
            # range parameter starts at zero, so just subtract 1 off of
            # the part number
            extra_part_args['CopySourceRange'] = calculate_range_parameter(
                part_size,
                part_number - 1,
                num_parts,
                transfer_future.meta.size,
            )
            # Get the size of the part copy as well for the progress
            # callbacks.
            size = self._get_transfer_size(
                part_size,
                part_number - 1,
                num_parts,
                transfer_future.meta.size,
            )
            part_futures.append(
                self._transfer_coordinator.submit(
                    request_executor,
                    CopyPartTask(
                        transfer_coordinator=self._transfer_coordinator,
                        main_kwargs={
                            'client': client,
                            'copy_source': call_args.copy_source,
                            'bucket': call_args.bucket,
                            'key': call_args.key,
                            'part_number': part_number,
                            'extra_args': extra_part_args,
                            'callbacks': progress_callbacks,
                            'size': size,
                        },
                        pending_main_kwargs={
                            'upload_id': create_multipart_future
                        },
                    ),
                )
            )

        complete_multipart_extra_args = self._extra_complete_multipart_args(
            call_args.extra_args
        )
        # Submit the request to complete the multipart upload.
        self._transfer_coordinator.submit(
            request_executor,
            CompleteMultipartUploadTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': complete_multipart_extra_args,
                },
                pending_main_kwargs={
                    'upload_id': create_multipart_future,
                    'parts': part_futures,
                },
                is_final=True,
            ),
        )

    def _get_head_object_request_from_copy_source(self, copy_source):
        if isinstance(copy_source, dict):
            return copy.copy(copy_source)
        else:
            raise TypeError(
                'Expecting dictionary formatted: '
                '{"Bucket": bucket_name, "Key": key} '
                'but got %s or type %s.' % (copy_source, type(copy_source))
            )

    def _extra_upload_part_args(self, extra_args):
        # Only the args in COPY_PART_ARGS actually need to be passed
        # onto the upload_part_copy calls.
        return get_filtered_dict(extra_args, self.UPLOAD_PART_COPY_ARGS)

    def _extra_complete_multipart_args(self, extra_args):
        return get_filtered_dict(extra_args, self.COMPLETE_MULTIPART_ARGS)

    def _get_transfer_size(
        self, part_size, part_index, num_parts, total_transfer_size
    ):
        if part_index == num_parts - 1:
            # The last part may be different in size then the rest of the
            # parts.
            return total_transfer_size - (part_index * part_size)
        return part_size class DeleteSubmissionTask(SubmissionTask):
    """Task for submitting tasks to execute an object deletion."""

    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

        self._transfer_coordinator.submit(
            request_executor,
            DeleteObjectTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': call_args.extra_args,
                },
                is_final=True,
            ),
        ) class DownloadSubmissionTask(SubmissionTask):
    """Task for submitting tasks to execute a download"""

    def _get_download_output_manager_cls(self, transfer_future, osutil):
        """Retrieves a class for managing output for a download

        :type transfer_future: s3transfer.futures.TransferFuture
        :param transfer_future: The transfer future for the request

        :type osutil: s3transfer.utils.OSUtils
        :param osutil: The os utility associated to the transfer

        :rtype: class of DownloadOutputManager
        :returns: The appropriate class to use for managing a specific type of
            input for downloads.
        """
        download_manager_resolver_chain = [
            DownloadSpecialFilenameOutputManager,
            DownloadFilenameOutputManager,
            DownloadSeekableOutputManager,
            DownloadNonSeekableOutputManager,
        ]

        fileobj = transfer_future.meta.call_args.fileobj
        for download_manager_cls in download_manager_resolver_chain:
            if download_manager_cls.is_compatible(fileobj, osutil):
                return download_manager_cls
        raise RuntimeError(
            'Output {} of type: {} is not supported.'.format(
                fileobj, type(fileobj)
            )
        )

    def _submit(
        self,
        client,
        config,
        osutil,
        request_executor,
        io_executor,
        transfer_future,
        bandwidth_limiter=None,
    ):
        """
        :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 io_executor: s3transfer.futures.BoundedExecutor
        :param io_executor: The io 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

        :type bandwidth_limiter: s3transfer.bandwidth.BandwidthLimiter
        :param bandwidth_limiter: The bandwidth limiter to use when
            downloading streams
        """
        if transfer_future.meta.size is None:
            # If a size was not provided figure out the size for the
            # user.
            response = client.head_object(
                Bucket=transfer_future.meta.call_args.bucket,
                Key=transfer_future.meta.call_args.key,
                **transfer_future.meta.call_args.extra_args,
            )
            transfer_future.meta.provide_transfer_size(
                response['ContentLength']
            )

        download_output_manager = self._get_download_output_manager_cls(
            transfer_future, osutil
        )(osutil, self._transfer_coordinator, io_executor)

        # If it is greater than threshold do a ranged download, otherwise
        # do a regular GetObject download.
        if transfer_future.meta.size < config.multipart_threshold:
            self._submit_download_request(
                client,
                config,
                osutil,
                request_executor,
                io_executor,
                download_output_manager,
                transfer_future,
                bandwidth_limiter,
            )
        else:
            self._submit_ranged_download_request(
                client,
                config,
                osutil,
                request_executor,
                io_executor,
                download_output_manager,
                transfer_future,
                bandwidth_limiter,
            )

    def _submit_download_request(
        self,
        client,
        config,
        osutil,
        request_executor,
        io_executor,
        download_output_manager,
        transfer_future,
        bandwidth_limiter,
    ):
        call_args = transfer_future.meta.call_args

        # Get a handle to the file that will be used for writing downloaded
        # contents
        fileobj = download_output_manager.get_fileobj_for_io_writes(
            transfer_future
        )

        # Get the needed callbacks for the task
        progress_callbacks = get_callbacks(transfer_future, 'progress')

        # Get any associated tags for the get object task.
        get_object_tag = download_output_manager.get_download_task_tag()

        # Get the final io task to run once the download is complete.
        final_task = download_output_manager.get_final_io_task()

        # Submit the task to download the object.
        self._transfer_coordinator.submit(
            request_executor,
            ImmediatelyWriteIOGetObjectTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'fileobj': fileobj,
                    'extra_args': call_args.extra_args,
                    'callbacks': progress_callbacks,
                    'max_attempts': config.num_download_attempts,
                    'download_output_manager': download_output_manager,
                    'io_chunksize': config.io_chunksize,
                    'bandwidth_limiter': bandwidth_limiter,
                },
                done_callbacks=[final_task],
            ),
            tag=get_object_tag,
        )

    def _submit_ranged_download_request(
        self,
        client,
        config,
        osutil,
        request_executor,
        io_executor,
        download_output_manager,
        transfer_future,
        bandwidth_limiter,
    ):
        call_args = transfer_future.meta.call_args

        # Get the needed progress callbacks for the task
        progress_callbacks = get_callbacks(transfer_future, 'progress')

        # Get a handle to the file that will be used for writing downloaded
        # contents
        fileobj = download_output_manager.get_fileobj_for_io_writes(
            transfer_future
        )

        # Determine the number of parts
        part_size = config.multipart_chunksize
        num_parts = calculate_num_parts(transfer_future.meta.size, part_size)

        # Get any associated tags for the get object task.
        get_object_tag = download_output_manager.get_download_task_tag()

        # Callback invoker to submit the final io task once all downloads
        # are complete.
        finalize_download_invoker = CountCallbackInvoker(
            self._get_final_io_task_submission_callback(
                download_output_manager, io_executor
            )
        )
        for i in range(num_parts):
            # Calculate the range parameter
            range_parameter = calculate_range_parameter(
                part_size, i, num_parts
            )

            # Inject the Range parameter to the parameters to be passed in
            # as extra args
            extra_args = {'Range': range_parameter}
            extra_args.update(call_args.extra_args)
            finalize_download_invoker.increment()
            # Submit the ranged downloads
            self._transfer_coordinator.submit(
                request_executor,
                GetObjectTask(
                    transfer_coordinator=self._transfer_coordinator,
                    main_kwargs={
                        'client': client,
                        'bucket': call_args.bucket,
                        'key': call_args.key,
                        'fileobj': fileobj,
                        'extra_args': extra_args,
                        'callbacks': progress_callbacks,
                        'max_attempts': config.num_download_attempts,
                        'start_index': i * part_size,
                        'download_output_manager': download_output_manager,
                        'io_chunksize': config.io_chunksize,
                        'bandwidth_limiter': bandwidth_limiter,
                    },
                    done_callbacks=[finalize_download_invoker.decrement],
                ),
                tag=get_object_tag,
            )
        finalize_download_invoker.finalize()

    def _get_final_io_task_submission_callback(
        self, download_manager, io_executor
    ):
        final_task = download_manager.get_final_io_task()
        return FunctionContainer(
            self._transfer_coordinator.submit, io_executor, final_task
        )

    def _calculate_range_param(self, part_size, part_index, num_parts):
        # Used to calculate the Range parameter
        start_range = part_index * part_size
        if part_index == num_parts - 1:
            end_range = ''
        else:
            end_range = start_range + part_size - 1
        range_param = f'bytes={start_range}-{end_range}'
        return range_param class RetriesExceededError(Exception):
class S3UploadFailedError(Exception):
class InvalidSubscriberMethodError(Exception):
class TransferNotDoneError(Exception):
class FatalError(CancelledError):
    def __init__(self, last_exception, msg='Max Retries Exceeded'): IN_MEMORY_DOWNLOAD_TAG = TaskTag('in_memory_download') IN_MEMORY_UPLOAD_TAG = TaskTag('in_memory_upload') class BoundedExecutor:
    EXECUTOR_CLS = futures.ThreadPoolExecutor

    def __init__(
        self, max_size, max_num_threads, tag_semaphores=None, executor_cls=None
    ):
        """An executor implementation that has a maximum queued up tasks

        The executor will block if the number of tasks that have been
        submitted and is currently working on is past its maximum.

        :params max_size: The maximum number of inflight futures. An inflight
            future means that the task is either queued up or is currently
            being executed. A size of None or 0 means that the executor will
            have no bound in terms of the number of inflight futures.

        :params max_num_threads: The maximum number of threads the executor
            uses.

        :type tag_semaphores: dict
        :params tag_semaphores: A dictionary where the key is the name of the
            tag and the value is the semaphore to use when limiting the
            number of tasks the executor is processing at a time.

        :type executor_cls: BaseExecutor
        :param underlying_executor_cls: The executor class that
            get bounded by this executor. If None is provided, the
            concurrent.futures.ThreadPoolExecutor class is used.
        """
        self._max_num_threads = max_num_threads
        if executor_cls is None:
            executor_cls = self.EXECUTOR_CLS
        self._executor = executor_cls(max_workers=self._max_num_threads)
        self._semaphore = TaskSemaphore(max_size)
        self._tag_semaphores = tag_semaphores

    def submit(self, task, tag=None, block=True):
        """Submit a task to complete

        :type task: s3transfer.tasks.Task
        :param task: The task to run __call__ on


        :type tag: s3transfer.futures.TaskTag
        :param tag: An optional tag to associate to the task. This
            is used to override which semaphore to use.

        :type block: boolean
        :param block: True if to wait till it is possible to submit a task.
            False, if not to wait and raise an error if not able to submit
            a task.

        :returns: The future associated to the submitted task
        """
        semaphore = self._semaphore
        # If a tag was provided, use the semaphore associated to that
        # tag.
        if tag:
            semaphore = self._tag_semaphores[tag]

        # Call acquire on the semaphore.
        acquire_token = semaphore.acquire(task.transfer_id, block)
        # Create a callback to invoke when task is done in order to call
        # release on the semaphore.
        release_callback = FunctionContainer(
            semaphore.release, task.transfer_id, acquire_token
        )
        # Submit the task to the underlying executor.
        future = ExecutorFuture(self._executor.submit(task))
        # Add the Semaphore.release() callback to the future such that
        # it is invoked once the future completes.
        future.add_done_callback(release_callback)
        return future

    def shutdown(self, wait=True):
        self._executor.shutdown(wait) class TransferCoordinator:
    """A helper class for managing TransferFuture"""

    def __init__(self, transfer_id=None):
        self.transfer_id = transfer_id
        self._status = 'not-started'
        self._result = None
        self._exception = None
        self._associated_futures = set()
        self._failure_cleanups = []
        self._done_callbacks = []
        self._done_event = threading.Event()
        self._lock = threading.Lock()
        self._associated_futures_lock = threading.Lock()
        self._done_callbacks_lock = threading.Lock()
        self._failure_cleanups_lock = threading.Lock()

    def __repr__(self):
        return '{}(transfer_id={})'.format(
            self.__class__.__name__, self.transfer_id
        )

    @property
    def exception(self):
        return self._exception

    @property
    def associated_futures(self):
        """The list of futures associated to the inprogress TransferFuture

        Once the transfer finishes this list becomes empty as the transfer
        is considered done and there should be no running futures left.
        """
        with self._associated_futures_lock:
            # We return a copy of the list because we do not want to
            # processing the returned list while another thread is adding
            # more futures to the actual list.
            return copy.copy(self._associated_futures)

    @property
    def failure_cleanups(self):
        """The list of callbacks to call when the TransferFuture fails"""
        return self._failure_cleanups

    @property
    def status(self):
        """The status of the TransferFuture

        The currently supported states are:
            * not-started - Has yet to start. If in this state, a transfer
              can be canceled immediately and nothing will happen.
            * queued - SubmissionTask is about to submit tasks
            * running - Is inprogress. In-progress as of now means that
              the SubmissionTask that runs the transfer is being executed. So
              there is no guarantee any transfer requests had been made to
              S3 if this state is reached.
            * cancelled - Was cancelled
            * failed - An exception other than CancelledError was thrown
            * success - No exceptions were thrown and is done.
        """
        return self._status

    def set_result(self, result):
        """Set a result for the TransferFuture

        Implies that the TransferFuture succeeded. This will always set a
        result because it is invoked on the final task where there is only
        ever one final task and it is ran at the very end of a transfer
        process. So if a result is being set for this final task, the transfer
        succeeded even if something came a long and canceled the transfer
        on the final task.
        """
        with self._lock:
            self._exception = None
            self._result = result
            self._status = 'success'

    def set_exception(self, exception, override=False):
        """Set an exception for the TransferFuture

        Implies the TransferFuture failed.

        :param exception: The exception that cause the transfer to fail.
        :param override: If True, override any existing state.
        """
        with self._lock:
            if not self.done() or override:
                self._exception = exception
                self._status = 'failed'

    def result(self):
        """Waits until TransferFuture is done and returns the result

        If the TransferFuture succeeded, it will return the result. If the
        TransferFuture failed, it will raise the exception associated to the
        failure.
        """
        # Doing a wait() with no timeout cannot be interrupted in python2 but
        # can be interrupted in python3 so we just wait with the largest
        # possible value integer value, which is on the scale of billions of
        # years...
        self._done_event.wait(MAXINT)

        # Once done waiting, raise an exception if present or return the
        # final result.
        if self._exception:
            raise self._exception
        return self._result

    def cancel(self, msg='', exc_type=CancelledError):
        """Cancels the TransferFuture

        :param msg: The message to attach to the cancellation
        :param exc_type: The type of exception to set for the cancellation
        """
        with self._lock:
            if not self.done():
                should_announce_done = False
                logger.debug('%s cancel(%s) called', self, msg)
                self._exception = exc_type(msg)
                if self._status == 'not-started':
                    should_announce_done = True
                self._status = 'cancelled'
                if should_announce_done:
                    self.announce_done()

    def set_status_to_queued(self):
        """Sets the TransferFutrue's status to running"""
        self._transition_to_non_done_state('queued')

    def set_status_to_running(self):
        """Sets the TransferFuture's status to running"""
        self._transition_to_non_done_state('running')

    def _transition_to_non_done_state(self, desired_state):
        with self._lock:
            if self.done():
                raise RuntimeError(
                    'Unable to transition from done state %s to non-done '
                    'state %s.' % (self.status, desired_state)
                )
            self._status = desired_state

    def submit(self, executor, task, tag=None):
        """Submits a task to a provided executor

        :type executor: s3transfer.futures.BoundedExecutor
        :param executor: The executor to submit the callable to

        :type task: s3transfer.tasks.Task
        :param task: The task to submit to the executor

        :type tag: s3transfer.futures.TaskTag
        :param tag: A tag to associate to the submitted task

        :rtype: concurrent.futures.Future
        :returns: A future representing the submitted task
        """
        logger.debug(
            "Submitting task {} to executor {} for transfer request: {}.".format(
                task, executor, self.transfer_id
            )
        )
        future = executor.submit(task, tag=tag)
        # Add this created future to the list of associated future just
        # in case it is needed during cleanups.
        self.add_associated_future(future)
        future.add_done_callback(
            FunctionContainer(self.remove_associated_future, future)
        )
        return future

    def done(self):
        """Determines if a TransferFuture has completed

        :returns: False if status is equal to 'failed', 'cancelled', or
            'success'. True, otherwise
        """
        return self.status in ['failed', 'cancelled', 'success']

    def add_associated_future(self, future):
        """Adds a future to be associated with the TransferFuture"""
        with self._associated_futures_lock:
            self._associated_futures.add(future)

    def remove_associated_future(self, future):
        """Removes a future's association to the TransferFuture"""
        with self._associated_futures_lock:
            self._associated_futures.remove(future)

    def add_done_callback(self, function, *args, **kwargs):
        """Add a done callback to be invoked when transfer is done"""
        with self._done_callbacks_lock:
            self._done_callbacks.append(
                FunctionContainer(function, *args, **kwargs)
            )

    def add_failure_cleanup(self, function, *args, **kwargs):
        """Adds a callback to call upon failure"""
        with self._failure_cleanups_lock:
            self._failure_cleanups.append(
                FunctionContainer(function, *args, **kwargs)
            )

    def announce_done(self):
        """Announce that future is done running and run associated callbacks

        This will run any failure cleanups if the transfer failed if not
        they have not been run, allows the result() to be unblocked, and will
        run any done callbacks associated to the TransferFuture if they have
        not already been ran.
        """
        if self.status != 'success':
            self._run_failure_cleanups()
        self._done_event.set()
        self._run_done_callbacks()

    def _run_done_callbacks(self):
        # Run the callbacks and remove the callbacks from the internal
        # list so they do not get ran again if done is announced more than
        # once.
        with self._done_callbacks_lock:
            self._run_callbacks(self._done_callbacks)
            self._done_callbacks = []

    def _run_failure_cleanups(self):
        # Run the cleanup callbacks and remove the callbacks from the internal
        # list so they do not get ran again if done is announced more than
        # once.
        with self._failure_cleanups_lock:
            self._run_callbacks(self.failure_cleanups)
            self._failure_cleanups = []

    def _run_callbacks(self, callbacks):
        for callback in callbacks:
            self._run_callback(callback)

    def _run_callback(self, callback):
        try:
            callback()
        # We do not want a callback interrupting the process, especially
        # in the failure cleanups. So log and catch, the exception.
        except Exception:
            logger.debug("Exception raised in %s." % callback, exc_info=True) class TransferFuture(BaseTransferFuture):
    def __init__(self, meta=None, coordinator=None):
        """The future associated to a submitted transfer request

        :type meta: TransferMeta
        :param meta: The metadata associated to the request. This object
            is visible to the requester.

        :type coordinator: TransferCoordinator
        :param coordinator: The coordinator associated to the request. This
            object is not visible to the requester.
        """
        self._meta = meta
        if meta is None:
            self._meta = TransferMeta()

        self._coordinator = coordinator
        if coordinator is None:
            self._coordinator = TransferCoordinator()

    @property
    def meta(self):
        return self._meta

    def done(self):
        return self._coordinator.done()

    def result(self):
        try:
            # Usually the result() method blocks until the transfer is done,
            # however if a KeyboardInterrupt is raised we want want to exit
            # out of this and propagate the exception.
            return self._coordinator.result()
        except KeyboardInterrupt as e:
            self.cancel()
            raise e

    def cancel(self):
        self._coordinator.cancel()

    def set_exception(self, exception):
        """Sets the exception on the future."""
        if not self.done():
            raise TransferNotDoneError(
                'set_exception can only be called once the transfer is '
                'complete.'
            )
        self._coordinator.set_exception(exception, override=True) class TransferMeta(BaseTransferMeta):
    """Holds metadata about the TransferFuture"""

    def __init__(self, call_args=None, transfer_id=None):
        self._call_args = call_args
        self._transfer_id = transfer_id
        self._size = None
        self._user_context = {}

    @property
    def call_args(self):
        """The call args used in the transfer request"""
        return self._call_args

    @property
    def transfer_id(self):
        """The unique id of the transfer"""
        return self._transfer_id

    @property
    def size(self):
        """The size of the transfer request if known"""
        return self._size

    @property
    def user_context(self):
        """A dictionary that requesters can store data in"""
        return self._user_context

    def provide_transfer_size(self, size):
        """A method to provide the size of a transfer request

        By providing this value, the TransferManager will not try to
        call HeadObject or use the use OS to determine the size of the
        transfer.
        """
        self._size = size class UploadSubmissionTask(SubmissionTask):
    """Task for submitting tasks to execute an upload"""

    UPLOAD_PART_ARGS = [
        'ChecksumAlgorithm',
        'SSECustomerKey',
        'SSECustomerAlgorithm',
        'SSECustomerKeyMD5',
        'RequestPayer',
        'ExpectedBucketOwner',
    ]

    COMPLETE_MULTIPART_ARGS = ['RequestPayer', 'ExpectedBucketOwner']

    def _get_upload_input_manager_cls(self, transfer_future):
        """Retrieves a class for managing input for an upload based on file type

        :type transfer_future: s3transfer.futures.TransferFuture
        :param transfer_future: The transfer future for the request

        :rtype: class of UploadInputManager
        :returns: The appropriate class to use for managing a specific type of
            input for uploads.
        """
        upload_manager_resolver_chain = [
            UploadFilenameInputManager,
            UploadSeekableInputManager,
            UploadNonSeekableInputManager,
        ]

        fileobj = transfer_future.meta.call_args.fileobj
        for upload_manager_cls in upload_manager_resolver_chain:
            if upload_manager_cls.is_compatible(fileobj):
                return upload_manager_cls
        raise RuntimeError(
            'Input {} of type: {} is not supported.'.format(
                fileobj, type(fileobj)
            )
        )

    def _submit(
        self,
        client,
        config,
        osutil,
        request_executor,
        transfer_future,
        bandwidth_limiter=None,
    ):
        """
        :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
        """
        upload_input_manager = self._get_upload_input_manager_cls(
            transfer_future
        )(osutil, self._transfer_coordinator, bandwidth_limiter)

        # Determine the size if it was not provided
        if transfer_future.meta.size is None:
            upload_input_manager.provide_transfer_size(transfer_future)

        # Do a multipart upload if needed, otherwise do a regular put object.
        if not upload_input_manager.requires_multipart_upload(
            transfer_future, config
        ):
            self._submit_upload_request(
                client,
                config,
                osutil,
                request_executor,
                transfer_future,
                upload_input_manager,
            )
        else:
            self._submit_multipart_request(
                client,
                config,
                osutil,
                request_executor,
                transfer_future,
                upload_input_manager,
            )

    def _submit_upload_request(
        self,
        client,
        config,
        osutil,
        request_executor,
        transfer_future,
        upload_input_manager,
    ):
        call_args = transfer_future.meta.call_args

        # Get any tags that need to be associated to the put object task
        put_object_tag = self._get_upload_task_tag(
            upload_input_manager, 'put_object'
        )

        # Submit the request of a single upload.
        self._transfer_coordinator.submit(
            request_executor,
            PutObjectTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'fileobj': upload_input_manager.get_put_object_body(
                        transfer_future
                    ),
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': call_args.extra_args,
                },
                is_final=True,
            ),
            tag=put_object_tag,
        )

    def _submit_multipart_request(
        self,
        client,
        config,
        osutil,
        request_executor,
        transfer_future,
        upload_input_manager,
    ):
        call_args = transfer_future.meta.call_args

        # Submit the request to create a multipart upload.
        create_multipart_future = self._transfer_coordinator.submit(
            request_executor,
            CreateMultipartUploadTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': call_args.extra_args,
                },
            ),
        )

        # Submit requests to upload the parts of the file.
        part_futures = []
        extra_part_args = self._extra_upload_part_args(call_args.extra_args)

        # Get any tags that need to be associated to the submitted task
        # for upload the data
        upload_part_tag = self._get_upload_task_tag(
            upload_input_manager, 'upload_part'
        )

        size = transfer_future.meta.size
        adjuster = ChunksizeAdjuster()
        chunksize = adjuster.adjust_chunksize(config.multipart_chunksize, size)
        part_iterator = upload_input_manager.yield_upload_part_bodies(
            transfer_future, chunksize
        )

        for part_number, fileobj in part_iterator:
            part_futures.append(
                self._transfer_coordinator.submit(
                    request_executor,
                    UploadPartTask(
                        transfer_coordinator=self._transfer_coordinator,
                        main_kwargs={
                            'client': client,
                            'fileobj': fileobj,
                            'bucket': call_args.bucket,
                            'key': call_args.key,
                            'part_number': part_number,
                            'extra_args': extra_part_args,
                        },
                        pending_main_kwargs={
                            'upload_id': create_multipart_future
                        },
                    ),
                    tag=upload_part_tag,
                )
            )

        complete_multipart_extra_args = self._extra_complete_multipart_args(
            call_args.extra_args
        )
        # Submit the request to complete the multipart upload.
        self._transfer_coordinator.submit(
            request_executor,
            CompleteMultipartUploadTask(
                transfer_coordinator=self._transfer_coordinator,
                main_kwargs={
                    'client': client,
                    'bucket': call_args.bucket,
                    'key': call_args.key,
                    'extra_args': complete_multipart_extra_args,
                },
                pending_main_kwargs={
                    'upload_id': create_multipart_future,
                    'parts': part_futures,
                },
                is_final=True,
            ),
        )

    def _extra_upload_part_args(self, extra_args):
        # Only the args in UPLOAD_PART_ARGS actually need to be passed
        # onto the upload_part calls.
        return get_filtered_dict(extra_args, self.UPLOAD_PART_ARGS)

    def _extra_complete_multipart_args(self, extra_args):
        return get_filtered_dict(extra_args, self.COMPLETE_MULTIPART_ARGS)

    def _get_upload_task_tag(self, upload_input_manager, operation_name):
        tag = None
        if upload_input_manager.stores_body_in_memory(operation_name):
            tag = IN_MEMORY_UPLOAD_TAG
        return tag class CallArgs:
    def __init__(self, **kwargs):
        """A class that records call arguments

        The call arguments must be passed as keyword arguments. It will set
        each keyword argument as an attribute of the object along with its
        associated value.
        """
        for arg, value in kwargs.items():
            setattr(self, arg, value) class OSUtils:
    _MAX_FILENAME_LEN = 255

    def get_file_size(self, filename):
        return os.path.getsize(filename)

    def open_file_chunk_reader(self, filename, start_byte, size, callbacks):
        return ReadFileChunk.from_filename(
            filename, start_byte, size, callbacks, enable_callbacks=False
        )

    def open_file_chunk_reader_from_fileobj(
        self,
        fileobj,
        chunk_size,
        full_file_size,
        callbacks,
        close_callbacks=None,
    ):
        return ReadFileChunk(
            fileobj,
            chunk_size,
            full_file_size,
            callbacks=callbacks,
            enable_callbacks=False,
            close_callbacks=close_callbacks,
        )

    def open(self, filename, mode):
        return open(filename, mode)

    def remove_file(self, filename):
        """Remove a file, noop if file does not exist."""
        # Unlike os.remove, if the file does not exist,
        # then this method does nothing.
        try:
            os.remove(filename)
        except OSError:
            pass

    def rename_file(self, current_filename, new_filename):
        rename_file(current_filename, new_filename)

    def is_special_file(cls, filename):
        """Checks to see if a file is a special UNIX file.

        It checks if the file is a character special device, block special
        device, FIFO, or socket.

        :param filename: Name of the file

        :returns: True if the file is a special file. False, if is not.
        """
        # If it does not exist, it must be a new file so it cannot be
        # a special file.
        if not os.path.exists(filename):
            return False
        mode = os.stat(filename).st_mode
        # Character special device.
        if stat.S_ISCHR(mode):
            return True
        # Block special device
        if stat.S_ISBLK(mode):
            return True
        # Named pipe / FIFO
        if stat.S_ISFIFO(mode):
            return True
        # Socket.
        if stat.S_ISSOCK(mode):
            return True
        return False

    def get_temp_filename(self, filename):
        suffix = os.extsep + random_file_extension()
        path = os.path.dirname(filename)
        name = os.path.basename(filename)
        temp_filename = name[: self._MAX_FILENAME_LEN - len(suffix)] + suffix
        return os.path.join(path, temp_filename)

    def allocate(self, filename, size):
        try:
            with self.open(filename, 'wb') as f:
                fallocate(f, size)
        except OSError:
            self.remove_file(filename)
            raise class SlidingWindowSemaphore(TaskSemaphore):
    """A semaphore used to coordinate sequential resource access.

    This class is similar to the stdlib BoundedSemaphore:

    * It's initialized with a count.
    * Each call to ``acquire()`` decrements the counter.
    * If the count is at zero, then ``acquire()`` will either block until the
      count increases, or if ``blocking=False``, then it will raise
      a NoResourcesAvailable exception indicating that it failed to acquire the
      semaphore.

    The main difference is that this semaphore is used to limit
    access to a resource that requires sequential access.  For example,
    if I want to access resource R that has 20 subresources R_0 - R_19,
    this semaphore can also enforce that you only have a max range of
    10 at any given point in time.  You must also specify a tag name
    when you acquire the semaphore.  The sliding window semantics apply
    on a per tag basis.  The internal count will only be incremented
    when the minimum sequence number for a tag is released.

    """

    def __init__(self, count):
        self._count = count
        # Dict[tag, next_sequence_number].
        self._tag_sequences = defaultdict(int)
        self._lowest_sequence = {}
        self._lock = threading.Lock()
        self._condition = threading.Condition(self._lock)
        # Dict[tag, List[sequence_number]]
        self._pending_release = {}

    def current_count(self):
        with self._lock:
            return self._count

    def acquire(self, tag, blocking=True):
        logger.debug("Acquiring %s", tag)
        self._condition.acquire()
        try:
            if self._count == 0:
                if not blocking:
                    raise NoResourcesAvailable("Cannot acquire tag '%s'" % tag)
                else:
                    while self._count == 0:
                        self._condition.wait()
            # self._count is no longer zero.
            # First, check if this is the first time we're seeing this tag.
            sequence_number = self._tag_sequences[tag]
            if sequence_number == 0:
                # First time seeing the tag, so record we're at 0.
                self._lowest_sequence[tag] = sequence_number
            self._tag_sequences[tag] += 1
            self._count -= 1
            return sequence_number
        finally:
            self._condition.release()

    def release(self, tag, acquire_token):
        sequence_number = acquire_token
        logger.debug("Releasing acquire %s/%s", tag, sequence_number)
        self._condition.acquire()
        try:
            if tag not in self._tag_sequences:
                raise ValueError("Attempted to release unknown tag: %s" % tag)
            max_sequence = self._tag_sequences[tag]
            if self._lowest_sequence[tag] == sequence_number:
                # We can immediately process this request and free up
                # resources.
                self._lowest_sequence[tag] += 1
                self._count += 1
                self._condition.notify()
                queued = self._pending_release.get(tag, [])
                while queued:
                    if self._lowest_sequence[tag] == queued[-1]:
                        queued.pop()
                        self._lowest_sequence[tag] += 1
                        self._count += 1
                    else:
                        break
            elif self._lowest_sequence[tag] < sequence_number < max_sequence:
                # We can't do anything right now because we're still waiting
                # for the min sequence for the tag to be released.  We have
                # to queue this for pending release.
                self._pending_release.setdefault(tag, []).append(
                    sequence_number
                )
                self._pending_release[tag].sort(reverse=True)
            else:
                raise ValueError(
                    "Attempted to release unknown sequence number "
                    "%s for tag: %s" % (sequence_number, tag)
                )
        finally:
            self._condition.release() class TaskSemaphore:
    def __init__(self, count):
        """A semaphore for the purpose of limiting the number of tasks

        :param count: The size of semaphore
        """
        self._semaphore = threading.Semaphore(count)

    def acquire(self, tag, blocking=True):
        """Acquire the semaphore

        :param tag: A tag identifying what is acquiring the semaphore. Note
            that this is not really needed to directly use this class but is
            needed for API compatibility with the SlidingWindowSemaphore
            implementation.
        :param block: If True, block until it can be acquired. If False,
            do not block and raise an exception if cannot be acquired.

        :returns: A token (can be None) to use when releasing the semaphore
        """
        logger.debug("Acquiring %s", tag)
        if not self._semaphore.acquire(blocking):
            raise NoResourcesAvailable("Cannot acquire tag '%s'" % tag)

    def release(self, tag, acquire_token):
        """Release the semaphore

        :param tag: A tag identifying what is releasing the semaphore
        :param acquire_token:  The token returned from when the semaphore was
            acquired. Note that this is not really needed to directly use this
            class but is needed for API compatibility with the
            SlidingWindowSemaphore implementation.
        """
        logger.debug(f"Releasing acquire {tag}/{acquire_token}")
        self._semaphore.release() def get_callbacks(transfer_future, callback_type):
    """Retrieves callbacks from a subscriber

    :type transfer_future: s3transfer.futures.TransferFuture
    :param transfer_future: The transfer future the subscriber is associated
        to.

    :type callback_type: str
    :param callback_type: The type of callback to retrieve from the subscriber.
        Valid types include:
            * 'queued'
            * 'progress'
            * 'done'

    :returns: A list of callbacks for the type specified. All callbacks are
        preinjected with the transfer future.
    """
    callbacks = []
    for subscriber in transfer_future.meta.call_args.subscribers:
        callback_name = 'on_' + callback_type
        if hasattr(subscriber, callback_name):
            callbacks.append(
                functools.partial(
                    getattr(subscriber, callback_name), future=transfer_future
                )
            )
    return callbacks def signal_not_transferring(request, operation_name, **kwargs):
    if operation_name in ['PutObject', 'UploadPart'] and hasattr(
        request.body, 'signal_not_transferring'
    ):
        request.body.signal_not_transferring() def signal_transferring(request, operation_name, **kwargs):
    if operation_name in ['PutObject', 'UploadPart'] and hasattr(
        request.body, 'signal_transferring'
    ):
        request.body.signal_transferring()
import copy
import logging
import re
import threading
from s3transfer.bandwidth import BandwidthLimiter, LeakyBucket
from s3transfer.constants import ALLOWED_DOWNLOAD_ARGS, KB, MB
from s3transfer.copies import CopySubmissionTask
from s3transfer.delete import DeleteSubmissionTask
from s3transfer.download import DownloadSubmissionTask
from s3transfer.exceptions import CancelledError, FatalError
from s3transfer.futures import (
    IN_MEMORY_DOWNLOAD_TAG,
    IN_MEMORY_UPLOAD_TAG,
    BoundedExecutor,
    TransferCoordinator,
    TransferFuture,
    TransferMeta,
)
from s3transfer.upload import UploadSubmissionTask
from s3transfer.utils import (
    CallArgs,
    OSUtils,
    SlidingWindowSemaphore,
    TaskSemaphore,
    get_callbacks,
    signal_not_transferring,
    signal_transferring,
)
            extra_args = {}
        if subscribers is None:
            subscribers = []
        self._validate_all_known_args(extra_args, self.ALLOWED_UPLOAD_ARGS)
        self._validate_if_bucket_supported(bucket)
        call_args = CallArgs(
            fileobj=fileobj,
            bucket=bucket,
            key=key,
            extra_args=extra_args,
            subscribers=subscribers,
        )
        extra_main_kwargs = {}
        if self._bandwidth_limiter:
            extra_main_kwargs['bandwidth_limiter'] = self._bandwidth_limiter
        return self._submit_transfer(
            call_args, UploadSubmissionTask, extra_main_kwargs
        )

    def download(
        self, bucket, key, fileobj, extra_args=None, subscribers=None
    ):
        """Downloads a file from S3

        :type bucket: str
        :param bucket: The name of the bucket to download from

        :type key: str
        :param key: The name of the key to download from

        :type fileobj: str or seekable file-like object
        :param fileobj: The name of a file to download or a seekable file-like
            object to download. It is recommended to use a filename because
            file-like objects may result in higher memory usage.

        :type extra_args: dict
        :param extra_args: Extra arguments that may be passed to the
            client operation

        :type subscribers: list(s3transfer.subscribers.BaseSubscriber)
        :param subscribers: The list of subscribers to be invoked in the
            order provided based on the event emit during the process of
            the transfer request.

        :rtype: s3transfer.futures.TransferFuture
        :returns: Transfer future representing the download
        """
        if extra_args is None:
            extra_args = {}
        if subscribers is None:
            subscribers = []
        self._validate_all_known_args(extra_args, self.ALLOWED_DOWNLOAD_ARGS)
        self._validate_if_bucket_supported(bucket)
        call_args = CallArgs(
            bucket=bucket,
            key=key,
            fileobj=fileobj,
            extra_args=extra_args,
            subscribers=subscribers,
        )
        extra_main_kwargs = {'io_executor': self._io_executor}
        if self._bandwidth_limiter:
            extra_main_kwargs['bandwidth_limiter'] = self._bandwidth_limiter
        return self._submit_transfer(
            call_args, DownloadSubmissionTask, extra_main_kwargs
        )

    def copy(
        self,
        copy_source,
        bucket,
        key,
        extra_args=None,
        subscribers=None,
        source_client=None,
    ):
        """Copies a file in S3

        :type copy_source: dict
        :param copy_source: The name of the source bucket, key name of the
            source object, and optional version ID of the source object. The
            dictionary format is:
            ``{'Bucket': 'bucket', 'Key': 'key', 'VersionId': 'id'}``. Note
            that the ``VersionId`` key is optional and may be omitted.

        :type bucket: str
        :param bucket: The name of the bucket to copy to

        :type key: str
        :param key: The name of the key to copy to

        :type extra_args: dict
        :param extra_args: Extra arguments that may be passed to the
            client operation

        :type subscribers: a list of subscribers
        :param subscribers: The list of subscribers to be invoked in the
            order provided based on the event emit during the process of
            the transfer request.

        :type source_client: botocore or boto3 Client
        :param source_client: The client to be used for operation that
            may happen at the source object. For example, this client is
            used for the head_object that determines the size of the copy.
            If no client is provided, the transfer manager's client is used
            as the client for the source object.

        :rtype: s3transfer.futures.TransferFuture
        :returns: Transfer future representing the copy
        """
        if extra_args is None:
            extra_args = {}
        if subscribers is None:
            subscribers = []
        if source_client is None:
            source_client = self._client
        self._validate_all_known_args(extra_args, self.ALLOWED_COPY_ARGS)
        if isinstance(copy_source, dict):
            self._validate_if_bucket_supported(copy_source.get('Bucket'))
        self._validate_if_bucket_supported(bucket)

        call_args = CallArgs(