python/alibabacloud_tea_openapi/client.py [136:229]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - self, action: str, version: str, protocol: str, method: str, auth_type: str, body_type: str, request: open_api_util_models.OpenApiRequest, runtime: RuntimeOptions, ) -> dict: _runtime = { 'key': runtime.key or self._key, 'cert': runtime.cert or self._cert, 'ca': runtime.ca or self._ca, 'readTimeout': DaraCore.to_number(runtime.read_timeout or self._read_timeout), 'connectTimeout': DaraCore.to_number(runtime.connect_timeout or self._connect_timeout), 'httpProxy': runtime.http_proxy or self._http_proxy, 'httpsProxy': runtime.https_proxy or self._https_proxy, 'noProxy': runtime.no_proxy or self._no_proxy, 'socks5Proxy': runtime.socks_5proxy or self._socks_5proxy, 'socks5NetWork': runtime.socks_5net_work or self._socks_5net_work, 'maxIdleConns': DaraCore.to_number(runtime.max_idle_conns or self._max_idle_conns), 'retryOptions': self._retry_options, 'ignoreSSL': runtime.ignore_ssl, 'tlsMinVersion': self._tls_min_version, } _last_request = None _last_response = None _retries_attempted = 0 _context = RetryPolicyContext( retries_attempted= _retries_attempted ) while DaraCore.should_retry(_runtime.get('retryOptions'), _context): if _retries_attempted > 0: _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context) if _backoff_time > 0: DaraCore.sleep(_backoff_time) _retries_attempted = _retries_attempted + 1 try: _request = DaraRequest() _request.protocol = self._protocol or protocol _request.method = method _request.pathname = '/' global_queries = {} global_headers = {} if not DaraCore.is_null(self._global_parameters): global_params = self._global_parameters if not DaraCore.is_null(global_params.queries): global_queries = global_params.queries if not DaraCore.is_null(global_params.headers): global_headers = global_params.headers extends_headers = {} extends_queries = {} if not DaraCore.is_null(runtime.extends_parameters): extends_parameters = runtime.extends_parameters if not DaraCore.is_null(extends_parameters.headers): extends_headers = extends_parameters.headers if not DaraCore.is_null(extends_parameters.queries): extends_queries = extends_parameters.queries _request.query = DaraCore.merge({ 'Action': action, 'Format': 'json', 'Version': version, 'Timestamp': Utils.get_timestamp(), 'SignatureNonce': Utils.get_nonce(), }, global_queries, extends_queries, request.query) headers = self.get_rpc_headers() if DaraCore.is_null(headers): # endpoint is setted in product client _request.headers = DaraCore.merge({ 'host': self._endpoint, 'x-acs-version': version, 'x-acs-action': action, 'user-agent': Utils.get_user_agent(self._user_agent), }, global_headers, extends_headers, request.headers) else: _request.headers = DaraCore.merge({ 'host': self._endpoint, 'x-acs-version': version, 'x-acs-action': action, 'user-agent': Utils.get_user_agent(self._user_agent), }, global_headers, extends_headers, request.headers, headers) if not DaraCore.is_null(request.body): m = request.body tmp = Utils.query(m) _request.body = DaraForm.to_form_string(tmp) _request.headers["content-type"] = 'application/x-www-form-urlencoded' if auth_type != 'Anonymous': if DaraCore.is_null(self._credential): raise main_exceptions.ClientException( code = f'InvalidCredentials', message = f'Please set up the credentials correctly. If you are setting them through environment variables, please ensure that ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set correctly. See https://help.aliyun.com/zh/sdk/developer-reference/configure-the-alibaba-cloud-accesskey-environment-variable-on-linux-macos-and-windows-systems for more details.' ) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - python/alibabacloud_tea_openapi/client.py [344:437]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - self, action: str, version: str, protocol: str, method: str, auth_type: str, body_type: str, request: open_api_util_models.OpenApiRequest, runtime: RuntimeOptions, ) -> dict: _runtime = { 'key': runtime.key or self._key, 'cert': runtime.cert or self._cert, 'ca': runtime.ca or self._ca, 'readTimeout': DaraCore.to_number(runtime.read_timeout or self._read_timeout), 'connectTimeout': DaraCore.to_number(runtime.connect_timeout or self._connect_timeout), 'httpProxy': runtime.http_proxy or self._http_proxy, 'httpsProxy': runtime.https_proxy or self._https_proxy, 'noProxy': runtime.no_proxy or self._no_proxy, 'socks5Proxy': runtime.socks_5proxy or self._socks_5proxy, 'socks5NetWork': runtime.socks_5net_work or self._socks_5net_work, 'maxIdleConns': DaraCore.to_number(runtime.max_idle_conns or self._max_idle_conns), 'retryOptions': self._retry_options, 'ignoreSSL': runtime.ignore_ssl, 'tlsMinVersion': self._tls_min_version, } _last_request = None _last_response = None _retries_attempted = 0 _context = RetryPolicyContext( retries_attempted= _retries_attempted ) while DaraCore.should_retry(_runtime.get('retryOptions'), _context): if _retries_attempted > 0: _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context) if _backoff_time > 0: DaraCore.sleep(_backoff_time) _retries_attempted = _retries_attempted + 1 try: _request = DaraRequest() _request.protocol = self._protocol or protocol _request.method = method _request.pathname = '/' global_queries = {} global_headers = {} if not DaraCore.is_null(self._global_parameters): global_params = self._global_parameters if not DaraCore.is_null(global_params.queries): global_queries = global_params.queries if not DaraCore.is_null(global_params.headers): global_headers = global_params.headers extends_headers = {} extends_queries = {} if not DaraCore.is_null(runtime.extends_parameters): extends_parameters = runtime.extends_parameters if not DaraCore.is_null(extends_parameters.headers): extends_headers = extends_parameters.headers if not DaraCore.is_null(extends_parameters.queries): extends_queries = extends_parameters.queries _request.query = DaraCore.merge({ 'Action': action, 'Format': 'json', 'Version': version, 'Timestamp': Utils.get_timestamp(), 'SignatureNonce': Utils.get_nonce(), }, global_queries, extends_queries, request.query) headers = self.get_rpc_headers() if DaraCore.is_null(headers): # endpoint is setted in product client _request.headers = DaraCore.merge({ 'host': self._endpoint, 'x-acs-version': version, 'x-acs-action': action, 'user-agent': Utils.get_user_agent(self._user_agent), }, global_headers, extends_headers, request.headers) else: _request.headers = DaraCore.merge({ 'host': self._endpoint, 'x-acs-version': version, 'x-acs-action': action, 'user-agent': Utils.get_user_agent(self._user_agent), }, global_headers, extends_headers, request.headers, headers) if not DaraCore.is_null(request.body): m = request.body tmp = Utils.query(m) _request.body = DaraForm.to_form_string(tmp) _request.headers["content-type"] = 'application/x-www-form-urlencoded' if auth_type != 'Anonymous': if DaraCore.is_null(self._credential): raise main_exceptions.ClientException( code = f'InvalidCredentials', message = f'Please set up the credentials correctly. If you are setting them through environment variables, please ensure that ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set correctly. See https://help.aliyun.com/zh/sdk/developer-reference/configure-the-alibaba-cloud-accesskey-environment-variable-on-linux-macos-and-windows-systems for more details.' ) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -