def append_object()

in python2/alibabacloud_oss_sdk/client.py [0:0]
99 lines of code
15 McCabe index (conditional complexity)

    def append_object(self, request, runtime):
        request.validate()
        runtime.validate()
        _runtime = {
            'timeouted': 'retry',
            'readTimeout': UtilClient.default_number(runtime.read_timeout, self._read_timeout),
            'connectTimeout': UtilClient.default_number(runtime.connect_timeout, self._connect_timeout),
            'localAddr': UtilClient.default_string(runtime.local_addr, self._local_addr),
            'httpProxy': UtilClient.default_string(runtime.http_proxy, self._http_proxy),
            'httpsProxy': UtilClient.default_string(runtime.https_proxy, self._https_proxy),
            'noProxy': UtilClient.default_string(runtime.no_proxy, self._no_proxy),
            'socks5Proxy': UtilClient.default_string(runtime.socks_5proxy, self._socks_5proxy),
            'socks5NetWork': UtilClient.default_string(runtime.socks_5net_work, self._socks_5net_work),
            'maxIdleConns': UtilClient.default_number(runtime.max_idle_conns, self._max_idle_conns),
            'retry': {
                'retryable': runtime.autoretry,
                'maxAttempts': UtilClient.default_number(runtime.max_attempts, 3)
            },
            'backoff': {
                'policy': UtilClient.default_string(runtime.backoff_policy, 'no'),
                'period': UtilClient.default_number(runtime.backoff_period, 1)
            },
            'ignoreSSL': runtime.ignore_ssl
        }
        _last_request = None
        _last_exception = None
        _now = time.time()
        _retry_times = 0
        while TeaCore.allow_retry(_runtime.get('retry'), _retry_times, _now):
            if _retry_times > 0:
                _backoff_time = TeaCore.get_backoff_time(_runtime.get('backoff'), _retry_times)
                if _backoff_time > 0:
                    TeaCore.sleep(_backoff_time)
            _retry_times = _retry_times + 1
            try:
                _request = TeaRequest()
                ctx = {}
                access_key_id = self._credential.get_access_key_id()
                access_key_secret = self._credential.get_access_key_secret()
                token = self._credential.get_security_token()
                _request.protocol = self._protocol
                _request.method = 'POST'
                _request.pathname = '/%s?append' % TeaConverter.to_unicode(request.object_name)
                _request.headers = TeaCore.merge({
                    'host': OSSUtilClient.get_host(request.bucket_name, self._region_id, self._endpoint, self._host_model),
                    'date': UtilClient.get_date_utcstring(),
                    'user-agent': self.get_user_agent()
                }, UtilClient.stringify_map_value(TeaCore.to_map(request.header)),
                    OSSUtilClient.parse_meta(request.user_meta, 'x-oss-meta-'))
                if not UtilClient.empty(token):
                    _request.headers['x-oss-security-token'] = token
                _request.query = UtilClient.stringify_map_value(TeaCore.to_map(request.filter))
                _request.body = OSSUtilClient.inject(request.body, ctx)
                if not UtilClient.is_unset(request.header) and not UtilClient.empty(request.header.content_type):
                    _request.headers['content-type'] = request.header.content_type
                else:
                    _request.headers['content-type'] = OSSUtilClient.get_content_type(request.object_name)
                _request.headers['authorization'] = OSSUtilClient.get_signature(_request, request.bucket_name, access_key_id, access_key_secret, self._signature_version, self._addtional_headers)
                _last_request = _request
                _response = TeaCore.do_action(_request, _runtime)
                resp_map = None
                body_str = None
                if UtilClient.is_4xx(_response.status_code) or UtilClient.is_5xx(_response.status_code):
                    body_str = UtilClient.read_as_string(_response.body)
                    resp_map = OSSUtilClient.get_err_message(body_str)
                    raise TeaException({
                        'code': resp_map.get('Code'),
                        'message': resp_map.get('Message'),
                        'data': {
                            'httpCode': _response.status_code,
                            'requestId': resp_map.get('RequestId'),
                            'hostId': resp_map.get('HostId')
                        }
                    })
                if self._is_enable_crc and not UtilClient.equal_string(ctx.get('crc'), _response.headers.get('x-oss-hash-crc64ecma')):
                    raise TeaException({
                        'code': 'CrcNotMatched',
                        'data': {
                            'clientCrc': ctx.get('crc'),
                            'serverCrc': _response.headers.get('x-oss-hash-crc64ecma')
                        }
                    })
                if self._is_enable_md5 and not UtilClient.equal_string(ctx.get('md5'), _response.headers.get('content-md5')):
                    raise TeaException({
                        'code': 'MD5NotMatched',
                        'data': {
                            'clientMD5': ctx.get('md5'),
                            'serverMD5': _response.headers.get('content-md5')
                        }
                    })
                return oss_models.AppendObjectResponse().from_map(
                    TeaCore.merge(_response.headers)
                )
            except Exception as e:
                if TeaCore.is_retryable(e):
                    _last_exception = e
                    continue
                raise e
        raise UnretryableException(_last_request, _last_exception)