def append_object()

in oss2/api.py [0:0]

• 20 lines of code
• 7 McCabe index (conditional complexity)

    def append_object(self, key, position, data,
                      headers=None,
                      progress_callback=None,
                      init_crc=None):
        """追加上传一个文件。

        :param str key: 新的文件名，或已经存在的可追加文件名
        :param int position: 追加上传一个新的文件， `position` 设为0；追加一个已经存在的可追加文件， `position` 设为文件的当前长度。
            `position` 可以从上次追加的结果 `AppendObjectResult.next_position` 中获得。

        :param data: 用户数据
        :type data: str、bytes、file-like object或可迭代对象

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

        :param progress_callback: 用户指定的进度回调函数。参考 :ref:`progress_callback`

        :return: :class:`AppendObjectResult <oss2.models.AppendObjectResult>`

        :raises: 如果 `position` 和当前文件长度不一致，抛出 :class:`PositionNotEqualToLength <oss2.exceptions.PositionNotEqualToLength>` ；
                 如果当前文件不是可追加类型，抛出 :class:`ObjectNotAppendable <oss2.exceptions.ObjectNotAppendable>` ；
                 还会抛出其他一些异常
        """
        headers = utils.set_content_type(http.CaseInsensitiveDict(headers), key)

        if progress_callback:
            data = utils.make_progress_adapter(data, progress_callback)

        if self.enable_crc and init_crc is not None:
            data = utils.make_crc_adapter(data, init_crc)

        logger.debug("Start to append object, bucket: {0}, key: {1}, headers: {2}, position: {3}".format(
            self.bucket_name, to_string(key), headers, position))
        resp = self.__do_object('POST', key,
                                data=data,
                                headers=headers,
                                params={'append': '', 'position': str(position)})
        logger.debug("Append object done, req_id: {0}, statu_code: {1}".format(resp.request_id, resp.status))
        result = AppendObjectResult(resp)

        if self.enable_crc and result.crc is not None and init_crc is not None:
            utils.check_crc('append object', data.crc, result.crc, result.request_id)

        return result