# coding: utf-8 from __future__ import absolute_import import datetime from multiprocessing.pool import ThreadPool from time import sleep # python 2 and python 3 compatibility library import six from volcenginesdkcore import rest from volcenginesdkcore.configuration import Configuration from volcenginesdkcore.interceptor import BuildRequestInterceptor, SignRequestInterceptor, \ ResolveEndpointInterceptor, RuntimeOptionsInterceptor from volcenginesdkcore.interceptor import DeserializedResponseInterceptor from volcenginesdkcore.interceptor import InterceptorChain, InterceptorContext from volcenginesdkcore.interceptor import Request, Response from volcenginesdkcore.observability.debugger import sdk_core_logger, LogLevel class ApiClient(object): """Generic API client for Swagger client library builds. Swagger generic API client. This client handles the client- server communication, and is invariant across implementations. Specifics of the methods and models for each application are generated from the Swagger templates. NOTE: This class is auto generated by the swagger code generator program. Ref: https://github.com/swagger-api/swagger-codegen Do not edit the class manually. :param configuration: .Configuration object for this client :param header_name: a header to pass when making calls to the API. :param header_value: a header value to pass when making calls to the API. :param cookie: a cookie to include in the header when making calls to the API """ PRIMITIVE_TYPES = (float, bool, bytes, six.text_type) + six.integer_types NATIVE_TYPES_MAPPING = { 'int': int, 'long': int if six.PY3 else long, # noqa: F821 'float': float, 'str': str, 'bool': bool, 'date': datetime.date, 'datetime': datetime.datetime, 'object': object, } def __init__(self, configuration=None, header_name=None, header_value=None, cookie=None): if configuration is None: configuration = Configuration() self.configuration = configuration # Use the pool property to lazily initialize the ThreadPool. self._pool = None self.rest_client = rest.RESTClientObject(configuration) self.default_headers = {} if header_name is not None: self.default_headers[header_name] = header_value self.cookie = cookie # Set default User-Agent. self.user_agent = 'volcstack-python-sdk/5.0.17' self.client_side_validation = configuration.client_side_validation self.interceptor_chain = InterceptorChain() self.interceptor_chain.append_request_interceptor(BuildRequestInterceptor()) self.interceptor_chain.append_request_interceptor(RuntimeOptionsInterceptor()) self.interceptor_chain.append_request_interceptor(ResolveEndpointInterceptor()) self.interceptor_chain.append_request_interceptor(SignRequestInterceptor()) self.interceptor_chain.append_response_interceptor(DeserializedResponseInterceptor()) def __del__(self): if self._pool is not None: self._pool.close() self._pool.join() @property def pool(self): if self._pool is None: self._pool = ThreadPool() return self._pool @property def user_agent(self): """User agent for this API client""" return self.default_headers['User-Agent'] @user_agent.setter def user_agent(self, value): self.default_headers['User-Agent'] = value def set_default_header(self, header_name, header_value): self.default_headers[header_name] = header_value def __call_api( self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None): # header parameters header_params = header_params or {} header_params.update(self.default_headers) if self.cookie: header_params['Cookie'] = self.cookie interceptor_context = InterceptorContext(request=Request( self.configuration, resource_path, method, path_params, query_params, header_params, body, post_params, files, response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout, )) interceptor_context = self.interceptor_chain.execute_request(interceptor_context) self._log_config(interceptor_context.get_request()) retry_count = 0 response_data = None retry_err = None auto_retry = interceptor_context.request.auto_retry retryer = interceptor_context.request.retryer num_max_retries = retryer.num_max_retries while True: # perform request and return response try: response_data = self.request( method, url=interceptor_context.request.url, query_params=interceptor_context.request.query_params, headers=interceptor_context.request.header_params, post_params=interceptor_context.request.post_params, body=interceptor_context.request.body, _preload_content=interceptor_context.request.preload_content, _request_timeout=interceptor_context.request.request_timeout) self.last_response = response_data except Exception as e: sdk_core_logger.warning("request error: {}".format(e)) retry_err = e if retry_count >= num_max_retries: raise e if not auto_retry or not self.request_retry(response_data, retry_count, retry_err, retryer): if retry_err is not None: raise retry_err break retry_count += 1 retry_err = None interceptor_context.response = Response(response_data) interceptor_context = self.interceptor_chain.execute_response(interceptor_context) response = interceptor_context.response if _return_http_data_only: return response.result else: return (response.result, response.http_response.status, response.http_response.getheaders()) def request_retry(self, response_data, retry_count, retry_err, retryer): if retryer.should_retry(response_data, retry_count, retry_err): delay = retryer.get_backoff_delay(retry_count) sleep(delay / 1000) if self.configuration.debug: sdk_core_logger.debug_config( "retry backoff strategy:%s, retry condition: %s, max retry count:%d, current retry count: %d, retry delay(ms):%f", type(retryer.backoff_strategy).__name__, type(retryer.retry_condition).__name__, retryer.num_max_retries, retry_count + 1, delay) return True return False def call_api(self, resource_path, method, path_params=None, query_params=None, header_params=None, body=None, post_params=None, files=None, response_type=None, auth_settings=None, async_req=None, _return_http_data_only=None, collection_formats=None, _preload_content=True, _request_timeout=None): """Makes the HTTP request (synchronous) and returns deserialized data. To make an async request, set the async_req parameter. :param resource_path: Path to method endpoint. :param method: Method to call. :param path_params: Path parameters in the url. :param query_params: Query parameters in the url. :param header_params: Header parameters to be placed in the request header. :param body: Request body. :param post_params dict: Request post form parameters, for `application/x-www-form-urlencoded`, `multipart/form-data`. :param auth_settings list: Auth Settings names for the request. :param response: Response data type. :param files dict: key -> filename, value -> filepath, for `multipart/form-data`. :param async_req bool: execute request asynchronously :param _return_http_data_only: response data without head status code and headers :param collection_formats: dict of collection formats for path, query, header, and post parameters. :param _preload_content: if False, the urllib3.HTTPResponse object will be returned without reading/decoding response data. Default is True. :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. :return: If async_req parameter is True, the request will be called asynchronously. The method will return the request thread. If parameter async_req is False or missing, then the method will return the response directly. """ if not async_req: return self.__call_api(resource_path, method, path_params, query_params, header_params, body, post_params, files, response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout) else: thread = self.pool.apply_async(self.__call_api, (resource_path, method, path_params, query_params, header_params, body, post_params, files, response_type, auth_settings, _return_http_data_only, collection_formats, _preload_content, _request_timeout)) return thread def request(self, method, url, query_params=None, headers=None, post_params=None, body=None, _preload_content=True, _request_timeout=None): """Makes the HTTP request using RESTClient.""" if method == "GET": return self.rest_client.GET(url, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers) elif method == "HEAD": return self.rest_client.HEAD(url, query_params=query_params, _preload_content=_preload_content, _request_timeout=_request_timeout, headers=headers) elif method == "OPTIONS": return self.rest_client.OPTIONS(url, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body) elif method == "POST": return self.rest_client.POST(url, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body) elif method == "PUT": return self.rest_client.PUT(url, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body) elif method == "PATCH": return self.rest_client.PATCH(url, query_params=query_params, headers=headers, post_params=post_params, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body) elif method == "DELETE": return self.rest_client.DELETE(url, query_params=query_params, headers=headers, _preload_content=_preload_content, _request_timeout=_request_timeout, body=body) else: raise ValueError( "http method must be `GET`, `HEAD`, `OPTIONS`," " `POST`, `PATCH`, `PUT` or `DELETE`." ) def select_header_accept(self, accepts): """Returns `Accept` based on an array of accepts provided. :param accepts: List of headers. :return: Accept (e.g. application/json). """ if not accepts: return accepts = [x.lower() for x in accepts] if 'application/json' in accepts: return 'application/json' else: return ', '.join(accepts) def select_header_content_type(self, content_types): """Returns `Content-Type` based on an array of content_types provided. :param content_types: List of content-types. :return: Content-Type (e.g. application/json). """ if not content_types: return 'application/json' content_types = [x.lower() for x in content_types] if 'application/json' in content_types or '*/*' in content_types: return 'application/json' else: return content_types[0] def _log_config(self, request): if not sdk_core_logger.is_enabled_for_loglevel(LogLevel.LOG_DEBUG_WITH_CONFIG): return sb = [] sb.append("SDK Version : ") sb.append(self.user_agent + "\n") # 连接池配置 sb.append("[Connection Pool]" + "\n") sb.append(" Number of pools : ") sb.append(str(self.configuration.num_pools) + "\n") sb.append(" Connection pool maxsize : ") sb.append(str(self.configuration.connection_pool_maxsize) + "\n") # SSL设置 sb.append("[Scheme]" + "\n") sb.append(" Scheme : ") sb.append(str(request.scheme) + "\n") # 代理设置（隐藏部分信息避免敏感泄露） sb.append("[Proxy]" + "\n"); sb.append(" HTTP Proxy : ") sb.append(str(self.configuration.http_proxy) + "\n") sb.append(" HTTPS Proxy : ") sb.append(str(self.configuration.https_proxy) + "\n") # 超时设置 sb.append("[Timeout]" + "\n"); sb.append(" Connect Timeout(ms) : ") sb.append(str(self.configuration.connect_timeout) + "\n") sb.append(" Read Timeout(ms) : ") sb.append(str(self.configuration.read_timeout) + "\n") # 重试设置 sb.append("[Retry]" + "\n"); sb.append(" Auto Retry : ") sb.append(str(request.auto_retry) + "\n") if request.auto_retry and request.retryer is not None: sb.append(" Max Retries : ") sb.append(str(request.retryer.num_max_retries) + "\n") sb.append(" Min Delay (ms) : ") sb.append(str(request.retryer.backoff_strategy.min_retry_delay_ms) + "\n") sb.append(" Max Delay (ms) : ") sb.append(str(request.retryer.backoff_strategy.max_retry_delay_ms) + "\n") sb.append(" Retry Condition : ") sb.append(type(request.retryer.retry_condition).__name__ if request.retryer.retry_condition is not None else "None" + "\n") sb.append(" Backoff Strategy : ") sb.append(type(request.retryer.backoff_strategy).__name__ if request.retryer.backoff_strategy is not None else "None" + "\n") sb.append(" Retry ErrorCodes : ") sb.append(str(request.retryer.retry_condition.retry_error_codes) + "\n") # EndpointResolver设置 sb.append("[Endpoint Resolver]" + "\n") sb.append(" Region : ") sb.append(str(request.region) + "\n") sb.append(" Endpoint : ") sb.append(str(request.host) + "\n") sb.append(" Use DualStack : ") sb.append(str(request.use_dual_stack) + "\n") sb.append(" Bootstrap Region : ") sb.append(str(request.custom_bootstrap_region) + "\n") sb.append(" Resolver Class : ") sb.append(type(request.endpoint_provider).__name__ if request.endpoint_provider is not None else "None" + "\n") sdk_core_logger.debug_config("".join(sb)) def metadata(self): return self._metadata