def upload_part()

in oss2/crypto_bucket.py [0:0]

• 25 lines of code
• 4 McCabe index (conditional complexity)

    def upload_part(self, key, upload_id, part_number, data, progress_callback=None, headers=None, upload_context=None):
        """客户端加密上传一个分片。

        :param upload_context:
        :param str key: 待上传文件名，这个文件名要和 :func:`init_multipart_upload` 的文件名一致。
        :param str upload_id: 分片上传ID
        :param int part_number: 分片号，最小值是1.
        :param data: 待上传数据。
        :param progress_callback: 用户指定进度回调函数。可以用来实现进度条等功能。参考 :ref:`progress_callback` 。

        :param headers: 用户指定的HTTP头部。可以指定Content-MD5头部等
        :type headers: 可以是dict，建议是oss2.CaseInsensitiveDict

        :return: :class:`PutObjectResult <oss2.models.PutObjectResult>`
        """
        logger.info(
            "Start to upload multipart of CryptoBucket, upload_id = {0}, part_number = {1}".format(upload_id,
                                                                                                   part_number))
        headers = http.CaseInsensitiveDict(headers)
        self._init_user_agent(headers)
        if upload_context:
            context = upload_context
        else:
            raise ClientError("Could not init upload context, upload contexts flag is False and upload context is none")

        content_crypto_material = context.content_crypto_material

        if content_crypto_material.cek_alg != self.crypto_provider.cipher.alg or content_crypto_material.wrap_alg != \
                self.crypto_provider.wrap_alg:
            err_msg = 'Envelope or data encryption/decryption algorithm is inconsistent'
            raise InconsistentError(err_msg, self)

        headers = content_crypto_material.to_object_meta(headers, context)

        plain_key = self.crypto_provider.decrypt_encrypted_key(content_crypto_material.encrypted_key)
        plain_iv = self.crypto_provider.decrypt_encrypted_iv(content_crypto_material.encrypted_iv)

        offset = context.part_size * (part_number - 1)
        counter = self.crypto_provider.cipher.calc_offset(offset)

        cipher = copy.copy(content_crypto_material.cipher)
        cipher.initialize(plain_key, plain_iv, counter)
        data = self.crypto_provider.make_encrypt_adapter(data, cipher)
        resp = super(CryptoBucket, self).upload_part(key, upload_id, part_number, data, progress_callback, headers)

        return resp