def readable(fileobj):
    """Determines whether or not a file-like object is readable.

    :param fileobj: The file-like object to determine if readable

    :returns: True, if readable. False otherwise.
    """
    if hasattr(fileobj, 'readable'):
        return fileobj.readable()

    return hasattr(fileobj, 'read') def seekable(fileobj):
    """Backwards compat function to determine if a fileobj is seekable

    :param fileobj: The file-like object to determine if seekable

    :returns: True, if seekable. False, otherwise.
    """
    # If the fileobj has a seekable attr, try calling the seekable()
    # method on it.
    if hasattr(fileobj, 'seekable'):
        return fileobj.seekable()
    # If there is no seekable attr, check if the object can be seeked
    # or telled. If it can, try to seek to the current position.
    elif hasattr(fileobj, 'seek') and hasattr(fileobj, 'tell'):
        try:
            fileobj.seek(0, 1)
            return True
        except OSError:
            # If an io related error was thrown then it is not seekable.
            return False
    # Else, the fileobj is not seekable
    return False IN_MEMORY_UPLOAD_TAG = TaskTag('in_memory_upload') class CompleteMultipartUploadTask(Task):
    """Task to complete a multipart upload"""

    def _main(self, client, bucket, key, upload_id, parts, extra_args):
        """
        :param client: The client to use when calling CompleteMultipartUpload
        :param bucket: The name of the bucket to upload to
        :param key: The name of the key to upload to
        :param upload_id: The id of the upload
        :param parts: A list of parts to use to complete the multipart upload::

            [{'Etag': etag_value, 'PartNumber': part_number}, ...]

            Each element in the list consists of a return value from
            ``UploadPartTask.main()``.
        :param extra_args:  A dictionary of any extra arguments that may be
            used in completing the multipart transfer.
        """
        client.complete_multipart_upload(
            Bucket=bucket,
            Key=key,
            UploadId=upload_id,
            MultipartUpload={'Parts': parts},
            **extra_args,
        ) class CreateMultipartUploadTask(Task):
    """Task to initiate a multipart upload"""

    def _main(self, client, bucket, key, extra_args):
        """
        :param client: The client to use when calling CreateMultipartUpload
        :param bucket: The name of the bucket to upload to
        :param key: The name of the key to upload to
        :param extra_args: A dictionary of any extra arguments that may be
            used in the initialization.

        :returns: The upload id of the multipart upload
        """
        # Create the multipart upload.
        response = client.create_multipart_upload(
            Bucket=bucket, Key=key, **extra_args
        )
        upload_id = response['UploadId']

        # Add a cleanup if the multipart upload fails at any point.
        self._transfer_coordinator.add_failure_cleanup(
            client.abort_multipart_upload,
            Bucket=bucket,
            Key=key,
            UploadId=upload_id,
        )
        return upload_id class SubmissionTask(Task):
    """A base class for any submission task

    Submission tasks are the top-level task used to submit a series of tasks
    to execute a particular transfer.
    """

    def _main(self, transfer_future, **kwargs):
        """
        :type transfer_future: s3transfer.futures.TransferFuture
        :param transfer_future: The transfer future associated with the
            transfer request that tasks are being submitted for

        :param kwargs: Any additional kwargs that you may want to pass
            to the _submit() method
        """
        try:
            self._transfer_coordinator.set_status_to_queued()

            # Before submitting any tasks, run all of the on_queued callbacks
            on_queued_callbacks = get_callbacks(transfer_future, 'queued')
            for on_queued_callback in on_queued_callbacks:
                on_queued_callback()

            # Once callbacks have been ran set the status to running.
            self._transfer_coordinator.set_status_to_running()

            # Call the submit method to start submitting tasks to execute the
            # transfer.
            self._submit(transfer_future=transfer_future, **kwargs)
        except BaseException as e:
            # If there was an exception raised during the submission of task
            # there is a chance that the final task that signals if a transfer
            # is done and too run the cleanup may never have been submitted in
            # the first place so we need to account accordingly.
            #
            # Note that BaseException is caught, instead of Exception, because
            # for some implementations of executors, specifically the serial
            # implementation, the SubmissionTask is directly exposed to
            # KeyboardInterupts and so needs to cleanup and signal done
            # for those as well.

            # Set the exception, that caused the process to fail.
            self._log_and_set_exception(e)

            # Wait for all possibly associated futures that may have spawned
            # from this submission task have finished before we announce the
            # transfer done.
            self._wait_for_all_submitted_futures_to_complete()

            # Announce the transfer as done, which will run any cleanups
            # and done callbacks as well.
            self._transfer_coordinator.announce_done()

    def _submit(self, transfer_future, **kwargs):
        """The submission method to be implemented

        :type transfer_future: s3transfer.futures.TransferFuture
        :param transfer_future: The transfer future associated with the
            transfer request that tasks are being submitted for

        :param kwargs: Any additional keyword arguments you want to be passed
            in
        """
        raise NotImplementedError('_submit() must be implemented')

    def _wait_for_all_submitted_futures_to_complete(self):
        # We want to wait for all futures that were submitted to
        # complete as we do not want the cleanup callbacks or done callbacks
        # to be called to early. The main problem is any task that was
        # submitted may have submitted even more during its process and so
        # we need to account accordingly.

        # First get all of the futures that were submitted up to this point.
        submitted_futures = self._transfer_coordinator.associated_futures
        while submitted_futures:
            # Wait for those futures to complete.
            self._wait_until_all_complete(submitted_futures)
            # However, more futures may have been submitted as we waited so
            # we need to check again for any more associated futures.
            possibly_more_submitted_futures = (
                self._transfer_coordinator.associated_futures
            )
            # If the current list of submitted futures is equal to the
            # the list of associated futures for when after the wait completes,
            # we can ensure no more futures were submitted in waiting on
            # the current list of futures to complete ultimately meaning all
            # futures that may have spawned from the original submission task
            # have completed.
            if submitted_futures == possibly_more_submitted_futures:
                break
            submitted_futures = possibly_more_submitted_futures class Task:
    """A task associated to a TransferFuture request

    This is a base class for other classes to subclass from. All subclassed
    classes must implement the main() method.
    """

    def __init__(
        self,
        transfer_coordinator,
        main_kwargs=None,
        pending_main_kwargs=None,
        done_callbacks=None,
        is_final=False,
    ):
        """
        :type transfer_coordinator: s3transfer.futures.TransferCoordinator
        :param transfer_coordinator: The context associated to the
            TransferFuture for which this Task is associated with.

        :type main_kwargs: dict
        :param main_kwargs: The keyword args that can be immediately supplied
            to the _main() method of the task

        :type pending_main_kwargs: dict
        :param pending_main_kwargs: The keyword args that are depended upon
            by the result from a dependent future(s). The result returned by
            the future(s) will be used as the value for the keyword argument
            when _main() is called. The values for each key can be:
                * a single future - Once completed, its value will be the
                  result of that single future
                * a list of futures - Once all of the futures complete, the
                  value used will be a list of each completed future result
                  value in order of when they were originally supplied.

        :type done_callbacks: list of callbacks
        :param done_callbacks: A list of callbacks to call once the task is
            done completing. Each callback will be called with no arguments
            and will be called no matter if the task succeeds or an exception
            is raised.

        :type is_final: boolean
        :param is_final: True, to indicate that this task is the final task
            for the TransferFuture request. By setting this value to True, it
            will set the result of the entire TransferFuture to the result
            returned by this task's main() method.
        """
        self._transfer_coordinator = transfer_coordinator

        self._main_kwargs = main_kwargs
        if self._main_kwargs is None:
            self._main_kwargs = {}

        self._pending_main_kwargs = pending_main_kwargs
        if pending_main_kwargs is None:
            self._pending_main_kwargs = {}

        self._done_callbacks = done_callbacks
        if self._done_callbacks is None:
            self._done_callbacks = []

        self._is_final = is_final

    def __repr__(self):
        # These are the general main_kwarg parameters that we want to
        # display in the repr.
        params_to_display = [
            'bucket',
            'key',
            'part_number',
            'final_filename',
            'transfer_future',
            'offset',
            'extra_args',
        ]
        main_kwargs_to_display = self._get_kwargs_with_params_to_include(
            self._main_kwargs, params_to_display
        )
        return '{}(transfer_id={}, {})'.format(
            self.__class__.__name__,
            self._transfer_coordinator.transfer_id,
            main_kwargs_to_display,
        )

    @property
    def transfer_id(self):
        """The id for the transfer request that the task belongs to"""
        return self._transfer_coordinator.transfer_id

    def _get_kwargs_with_params_to_include(self, kwargs, include):
        filtered_kwargs = {}
        for param in include:
            if param in kwargs:
                filtered_kwargs[param] = kwargs[param]
        return filtered_kwargs

    def _get_kwargs_with_params_to_exclude(self, kwargs, exclude):
        filtered_kwargs = {}
        for param, value in kwargs.items():
            if param in exclude:
                continue
            filtered_kwargs[param] = value
        return filtered_kwargs

    def __call__(self):
        """The callable to use when submitting a Task to an executor"""
        try:
            # Wait for all of futures this task depends on.
            self._wait_on_dependent_futures()
            # Gather up all of the main keyword arguments for main().
            # This includes the immediately provided main_kwargs and
            # the values for pending_main_kwargs that source from the return
            # values from the task's dependent futures.
            kwargs = self._get_all_main_kwargs()
            # If the task is not done (really only if some other related
            # task to the TransferFuture had failed) then execute the task's
            # main() method.
            if not self._transfer_coordinator.done():
                return self._execute_main(kwargs)
        except Exception as e:
            self._log_and_set_exception(e)
        finally:
            # Run any done callbacks associated to the task no matter what.
            for done_callback in self._done_callbacks:
                done_callback()

            if self._is_final:
                # If this is the final task announce that it is done if results
                # are waiting on its completion.
                self._transfer_coordinator.announce_done()

    def _execute_main(self, kwargs):
        # Do not display keyword args that should not be printed, especially
        # if they are going to make the logs hard to follow.
        params_to_exclude = ['data']
        kwargs_to_display = self._get_kwargs_with_params_to_exclude(
            kwargs, params_to_exclude
        )
        # Log what is about to be executed.
        logger.debug(f"Executing task {self} with kwargs {kwargs_to_display}")

        return_value = self._main(**kwargs)
        # If the task is the final task, then set the TransferFuture's
        # value to the return value from main().
        if self._is_final:
            self._transfer_coordinator.set_result(return_value)
        return return_value

    def _log_and_set_exception(self, exception):
        # If an exception is ever thrown than set the exception for the
        # entire TransferFuture.
        logger.debug("Exception raised.", exc_info=True)
        self._transfer_coordinator.set_exception(exception)

    def _main(self, **kwargs):
        """The method that will be ran in the executor

        This method must be implemented by subclasses from Task. main() can
        be implemented with any arguments decided upon by the subclass.
        """
        raise NotImplementedError('_main() must be implemented')

    def _wait_on_dependent_futures(self):
        # Gather all of the futures into that main() depends on.
        futures_to_wait_on = []
        for _, future in self._pending_main_kwargs.items():
            # If the pending main keyword arg is a list then extend the list.
            if isinstance(future, list):
                futures_to_wait_on.extend(future)
            # If the pending main keyword arg is a future append it to the list.
            else:
                futures_to_wait_on.append(future)
        # Now wait for all of the futures to complete.
        self._wait_until_all_complete(futures_to_wait_on)

    def _wait_until_all_complete(self, futures):
        # This is a basic implementation of the concurrent.futures.wait()
        #
        # concurrent.futures.wait() is not used instead because of this
        # reported issue: https://bugs.python.org/issue20319.
        # The issue would occasionally cause multipart uploads to hang
        # when wait() was called. With this approach, it avoids the
        # concurrency bug by removing any association with concurrent.futures
        # implementation of waiters.
        logger.debug(
            '%s about to wait for the following futures %s', self, futures
        )
        for future in futures:
            try:
                logger.debug('%s about to wait for %s', self, future)
                future.result()
            except Exception:
                # result() can also produce exceptions. We want to ignore
                # these to be deferred to error handling down the road.
                pass
        logger.debug('%s done waiting for dependent futures', self)

    def _get_all_main_kwargs(self):
        # Copy over all of the kwargs that we know is available.
        kwargs = copy.copy(self._main_kwargs)

        # Iterate through the kwargs whose values are pending on the result
        # of a future.
        for key, pending_value in self._pending_main_kwargs.items():
            # If the value is a list of futures, iterate though the list
            # appending on the result from each future.
            if isinstance(pending_value, list):
                result = []
                for future in pending_value:
                    result.append(future.result())
            # Otherwise if the pending_value is a future, just wait for it.
            else:
                result = pending_value.result()
            # Add the retrieved value to the kwargs to be sent to the
            # main() call.
            kwargs[key] = result
        return kwargs class ChunksizeAdjuster:
    def __init__(
        self,
        max_size=MAX_SINGLE_UPLOAD_SIZE,
        min_size=MIN_UPLOAD_CHUNKSIZE,
        max_parts=MAX_PARTS,
    ):
        self.max_size = max_size
        self.min_size = min_size
        self.max_parts = max_parts

    def adjust_chunksize(self, current_chunksize, file_size=None):
        """Get a chunksize close to current that fits within all S3 limits.

        :type current_chunksize: int
        :param current_chunksize: The currently configured chunksize.

        :type file_size: int or None
        :param file_size: The size of the file to upload. This might be None
            if the object being transferred has an unknown size.

        :returns: A valid chunksize that fits within configured limits.
        """
        chunksize = current_chunksize
        if file_size is not None:
            chunksize = self._adjust_for_max_parts(chunksize, file_size)
        return self._adjust_for_chunksize_limits(chunksize)

    def _adjust_for_chunksize_limits(self, current_chunksize):
        if current_chunksize > self.max_size:
            logger.debug(
                "Chunksize greater than maximum chunksize. "
                "Setting to %s from %s." % (self.max_size, current_chunksize)
            )
            return self.max_size
        elif current_chunksize < self.min_size:
            logger.debug(
                "Chunksize less than minimum chunksize. "
                "Setting to %s from %s." % (self.min_size, current_chunksize)
            )
            return self.min_size
        else:
            return current_chunksize

    def _adjust_for_max_parts(self, current_chunksize, file_size):
        chunksize = current_chunksize
        num_parts = int(math.ceil(file_size / float(chunksize)))

        while num_parts > self.max_parts:
            chunksize *= 2
            num_parts = int(math.ceil(file_size / float(chunksize)))

        if chunksize != current_chunksize:
            logger.debug(
                "Chunksize would result in the number of parts exceeding the "
                "maximum. Setting to %s from %s."
                % (chunksize, current_chunksize)
            )

        return chunksize class DeferredOpenFile:
    def __init__(self, filename, start_byte=0, mode='rb', open_function=open):
        """A class that defers the opening of a file till needed

        This is useful for deferring opening of a file till it is needed
        in a separate thread, as there is a limit of how many open files
        there can be in a single thread for most operating systems. The
        file gets opened in the following methods: ``read()``, ``seek()``,
        and ``__enter__()``

        :type filename: str
        :param filename: The name of the file to open

        :type start_byte: int
        :param start_byte: The byte to seek to when the file is opened.

        :type mode: str
        :param mode: The mode to use to open the file

        :type open_function: function
        :param open_function: The function to use to open the file
        """
        self._filename = filename
        self._fileobj = None
        self._start_byte = start_byte
        self._mode = mode
        self._open_function = open_function

    def _open_if_needed(self):
        if self._fileobj is None:
            self._fileobj = self._open_function(self._filename, self._mode)
            if self._start_byte != 0:
                self._fileobj.seek(self._start_byte)

    @property
    def name(self):
        return self._filename

    def read(self, amount=None):
        self._open_if_needed()
        return self._fileobj.read(amount)

    def write(self, data):
        self._open_if_needed()
        self._fileobj.write(data)

    def seek(self, where, whence=0):
        self._open_if_needed()
        self._fileobj.seek(where, whence)

    def tell(self):
        if self._fileobj is None:
            return self._start_byte
        return self._fileobj.tell()

    def close(self):
        if self._fileobj:
            self._fileobj.close()

    def __enter__(self):
        self._open_if_needed()
        return self

    def __exit__(self, *args, **kwargs):
        self.close() 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 get_filtered_dict(original_dict, whitelisted_keys):
    """Gets a dictionary filtered by whitelisted keys

    :param original_dict: The original dictionary of arguments to source keys
        and values.
    :param whitelisted_key: A list of keys to include in the filtered
        dictionary.

    :returns: A dictionary containing key/values from the original dictionary
        whose key was included in the whitelist
    """
    filtered_dict = {}
    for key, value in original_dict.items():
        if key in whitelisted_keys:
            filtered_dict[key] = value
    return filtered_dict
import math
from io import BytesIO
from s3transfer.compat import readable, seekable
from s3transfer.futures import IN_MEMORY_UPLOAD_TAG
from s3transfer.tasks import (
    CompleteMultipartUploadTask,
    CreateMultipartUploadTask,
    SubmissionTask,
    Task,
)
from s3transfer.utils import (
    ChunksizeAdjuster,
    DeferredOpenFile,
    get_callbacks,
    get_filtered_dict,
)
        :type config: s3transfer.manager.TransferConfig
        :param config: The config associated to the transfer manager

        :rtype: s3transfer.utils.ReadFileChunk
        :returns: A ReadFileChunk including all progress callbacks
            associated with the transfer future.
        """
        raise NotImplementedError('must implement get_put_object_body()')

    def yield_upload_part_bodies(self, transfer_future, chunksize):
        """Yields the part number and body to use for each UploadPart

        :type transfer_future: s3transfer.futures.TransferFuture
        :param transfer_future: The future associated with upload request

        :type chunksize: int
        :param chunksize: The chunksize to use for this upload.

        :rtype: int, s3transfer.utils.ReadFileChunk
        :returns: Yields the part number and the ReadFileChunk including all
            progress callbacks associated with the transfer future for that
            specific yielded part.
        """
        raise NotImplementedError('must implement yield_upload_part_bodies()')

    def _wrap_fileobj(self, fileobj):
        fileobj = InterruptReader(fileobj, self._transfer_coordinator)
        if self._bandwidth_limiter:
            fileobj = self._bandwidth_limiter.get_bandwith_limited_stream(
                fileobj, self._transfer_coordinator, enabled=False
            )
        return fileobj

    def _get_progress_callbacks(self, transfer_future):
        callbacks = get_callbacks(transfer_future, 'progress')
        # We only want to be wrapping the callbacks if there are callbacks to
        # invoke because we do not want to be doing any unnecessary work if
        # there are no callbacks to invoke.
        if callbacks:
            return [AggregatedProgressCallback(callbacks)]
        return []

    def _get_close_callbacks(self, aggregated_progress_callbacks):
        return [callback.flush for callback in aggregated_progress_callbacks]


class UploadFilenameInputManager(UploadInputManager):
    """Upload utility for filenames"""

    @classmethod
    def is_compatible(cls, upload_source):
        return isinstance(upload_source, str)

    def stores_body_in_memory(self, operation_name):
        return False

    def provide_transfer_size(self, transfer_future):
        transfer_future.meta.provide_transfer_size(
            self._osutil.get_file_size(transfer_future.meta.call_args.fileobj)
        )

    def requires_multipart_upload(self, transfer_future, config):
        return transfer_future.meta.size >= config.multipart_threshold

    def get_put_object_body(self, transfer_future):
        # Get a file-like object for the given input
        fileobj, full_size = self._get_put_object_fileobj_with_full_size(
            transfer_future
        )

        # Wrap fileobj with interrupt reader that will quickly cancel
        # uploads if needed instead of having to wait for the socket
        # to completely read all of the data.
        fileobj = self._wrap_fileobj(fileobj)

        callbacks = self._get_progress_callbacks(transfer_future)
        close_callbacks = self._get_close_callbacks(callbacks)
        size = transfer_future.meta.size
        # Return the file-like object wrapped into a ReadFileChunk to get
        # progress.
        return self._osutil.open_file_chunk_reader_from_fileobj(
            fileobj=fileobj,
            chunk_size=size,
            full_file_size=full_size,
            callbacks=callbacks,
            close_callbacks=close_callbacks,
        )

    def yield_upload_part_bodies(self, transfer_future, chunksize):
        full_file_size = transfer_future.meta.size
        num_parts = self._get_num_parts(transfer_future, chunksize)
        for part_number in range(1, num_parts + 1):
            callbacks = self._get_progress_callbacks(transfer_future)
            close_callbacks = self._get_close_callbacks(callbacks)
            start_byte = chunksize * (part_number - 1)
            # Get a file-like object for that part and the size of the full
            # file size for the associated file-like object for that part.
            fileobj, full_size = self._get_upload_part_fileobj_with_full_size(
                transfer_future.meta.call_args.fileobj,
                start_byte=start_byte,
                part_size=chunksize,
                full_file_size=full_file_size,
            )

            # Wrap fileobj with interrupt reader that will quickly cancel
            # uploads if needed instead of having to wait for the socket
            # to completely read all of the data.
            fileobj = self._wrap_fileobj(fileobj)

            # Wrap the file-like object into a ReadFileChunk to get progress.
            read_file_chunk = self._osutil.open_file_chunk_reader_from_fileobj(
                fileobj=fileobj,
                chunk_size=chunksize,
                full_file_size=full_size,
                callbacks=callbacks,
                close_callbacks=close_callbacks,
            )
            yield part_number, read_file_chunk

    def _get_deferred_open_file(self, fileobj, start_byte):

        fileobj = DeferredOpenFile(