Class UploadPartTask

awscli/s3transfer/upload.py:803–844

Task to upload a part in a multipart upload

class UploadPartTask(Task):
    """Task to upload a part in a multipart upload"""

    def _main(
        self, client, fileobj, bucket, key, upload_id, part_number, extra_args
    ):
        """
        :param client: The client to use when calling PutObject
        :param fileobj: The file to upload.
        :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 part_number: The number representing the part of the multipart
            upload
        :param extra_args: A dictionary of any extra arguments that may be
            used in the upload.

        :rtype: dict
        :returns: A dictionary representing a part::

            {'Etag': etag_value, 'PartNumber': part_number}

            This value can be appended to a list to be used to complete
            the multipart upload.
        """
        with fileobj as body:
            response = client.upload_part(
                Bucket=bucket,
                Key=key,
                UploadId=upload_id,
                PartNumber=part_number,
                Body=body,
                **extra_args,
            )
        etag = response['ETag']
        part_metadata = {'ETag': etag, 'PartNumber': part_number}
        if 'ChecksumAlgorithm' in extra_args:
            algorithm_name = extra_args['ChecksumAlgorithm'].upper()
            checksum_member = f'Checksum{algorithm_name}'
            if checksum_member in response:
                part_metadata[checksum_member] = response[checksum_member]
        return part_metadata