def get_object_with_url()

in oss2/crypto_bucket.py [0:0]

• 29 lines of code
• 9 McCabe index (conditional complexity)

    def get_object_with_url(self, sign_url,
                            byte_range=None,
                            headers=None,
                            progress_callback=None):
        """使用加签的url下载文件

        :param sign_url: 加签的url
        :param byte_range: 指定下载范围。参见 :ref:`byte_range`

        :param headers: HTTP头部
        :type headers: 可以是dict，建议是oss2.CaseInsensitiveDict，必须和签名时保持一致

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

        :return: file-like object

        :raises: 如果文件不存在，则抛出 :class:`NoSuchKey <oss2.exceptions.NoSuchKey>` ；还可能抛出其他异常
        """
        query = parse_qs(urlsplit(sign_url).query)
        if query and (Bucket.PROCESS in query):
            raise ClientError("Process object operation is not support for Crypto Bucket")

        headers = http.CaseInsensitiveDict(headers)
        self._init_user_agent(headers)

        discard = 0
        range_string = ''

        if byte_range:
            if not byte_range[0] and byte_range[1]:
                raise ClientError("Don't support range get while start is none and end is not")
            start, end = self.crypto_provider.adjust_range(byte_range[0], byte_range[1])
            adjust_byte_range = (start, end)

            range_string = _make_range_string(adjust_byte_range)
            if range_string:
                headers['range'] = range_string

            if byte_range[0] and adjust_byte_range[0] < byte_range[0]:
                discard = byte_range[0] - adjust_byte_range[0]
            logger.debug("adjust range of get object, byte_range: {0}, adjust_byte_range: {1}, discard: {2}".format(
                byte_range, adjust_byte_range, discard))

        logger.debug(
            "Start to get object with url from CryptoBucket: {0}, sign_url: {1}, range: {2}, headers: {3}".format(
                self.bucket_name, sign_url, range_string, headers))
        resp = self._do_url('GET', sign_url, headers=headers)
        return GetObjectResult(resp, progress_callback, self.enable_crc,
                               crypto_provider=self.crypto_provider, discard=discard)