shanxia2.0/backend/venv/lib/python3.9/site-packages/alibabacloud_tea_openapi/client.py

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
from __future__ import annotations

import os

from typing import Dict, Generator, AsyncGenerator, Any

from alibabacloud_credentials import models as credential_models
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_gateway_spi import models as spi_models
from alibabacloud_gateway_spi.client import Client as SPIClient
from alibabacloud_tea_openapi import exceptions as main_exceptions
from alibabacloud_tea_openapi import models as main_models
from alibabacloud_tea_openapi import utils_models as open_api_util_models
from alibabacloud_tea_openapi.utils import Utils
from darabonba.core import DaraCore as DaraCore
from darabonba.core import DaraCore
from darabonba.exceptions import DaraException, UnretryableException
from darabonba.policy.retry import RetryOptions, RetryPolicyContext
from darabonba.request import DaraRequest
from darabonba.runtime import RuntimeOptions
from darabonba.utils.bytes import Bytes as DaraBytes
from darabonba.utils.form import Form as DaraForm
from darabonba.utils.stream import Stream as DaraStream
from darabonba.utils.xml import XML as DaraXML
from darabonba.websocket import (
    get_web_socket_enable_reconnect,
    get_web_socket_handshake_timeout,
    get_web_socket_max_reconnect_times,
    get_web_socket_ping_interval,
    get_web_socket_pong_timeout,
    get_web_socket_reconnect_interval,
    get_web_socket_write_timeout,
    new_websocket_client_and_connect,
)
from alibabacloud_tea_openapi import websocket_utils

"""
 * @remarks
 * This is for OpenApi SDK
"""
class Client:
    _endpoint: str = None
    _region_id: str = None
    _protocol: str = None
    _method: str = None
    _user_agent: str = None
    _endpoint_rule: str = None
    _endpoint_map: Dict[str, str] = None
    _suffix: str = None
    _read_timeout: int = None
    _connect_timeout: int = None
    _http_proxy: str = None
    _https_proxy: str = None
    _socks_5proxy: str = None
    _socks_5net_work: str = None
    _no_proxy: str = None
    _network: str = None
    _product_id: str = None
    _max_idle_conns: int = None
    _endpoint_type: str = None
    _open_platform_endpoint: str = None
    _credential: CredentialClient = None
    _signature_version: str = None
    _signature_algorithm: str = None
    _headers: Dict[str, str] = None
    _spi: SPIClient = None
    _global_parameters: open_api_util_models.GlobalParameters = None
    _key: str = None
    _cert: str = None
    _ca: str = None
    _disable_http_2: bool = None
    _retry_options: RetryOptions = None
    _tls_min_version: str = None
    _attribute_map: spi_models.AttributeMap = None

    def __init__(
        self,
        config: open_api_util_models.Config,
    ):
        if DaraCore.is_null(config):
            raise main_exceptions.ClientException(
                code = 'ParameterMissing',
                message = '\'config\' can not be unset'
            )
        if (not DaraCore.is_null(config.access_key_id) and config.access_key_id != '') and (not DaraCore.is_null(config.access_key_secret) and config.access_key_secret != ''):
            if not DaraCore.is_null(config.security_token) and config.security_token != '':
                config.type = 'sts'
            else:
                config.type = 'access_key'

            credential_config = credential_models.Config(
                access_key_id = config.access_key_id,
                type = config.type,
                access_key_secret = config.access_key_secret
            )
            credential_config.security_token = config.security_token
            self._credential = CredentialClient(credential_config)
        elif not DaraCore.is_null(config.bearer_token) and config.bearer_token != '':
            cc = credential_models.Config(
                type = 'bearer',
                bearer_token = config.bearer_token
            )
            self._credential = CredentialClient(cc)
        elif not DaraCore.is_null(config.credential):
            self._credential = config.credential
        if DaraCore.is_null(config.user_agent):
            self._user_agent = os.environ.get('ALIBABA_CLOUD_USER_AGENT')
        else:
            self._user_agent = config.user_agent

        self._endpoint = config.endpoint
        self._endpoint_type = config.endpoint_type
        self._network = config.network
        self._suffix = config.suffix
        self._protocol = config.protocol
        self._method = config.method
        self._region_id = config.region_id
        self._read_timeout = config.read_timeout
        self._connect_timeout = config.connect_timeout
        self._http_proxy = config.http_proxy
        self._https_proxy = config.https_proxy
        self._no_proxy = config.no_proxy
        self._socks_5proxy = config.socks_5proxy
        self._socks_5net_work = config.socks_5net_work
        self._max_idle_conns = config.max_idle_conns
        self._signature_version = config.signature_version
        self._signature_algorithm = config.signature_algorithm
        self._global_parameters = config.global_parameters
        self._key = config.key
        self._cert = config.cert
        self._ca = config.ca
        self._disable_http_2 = config.disable_http_2
        self._retry_options = config.retry_options
        self._tls_min_version = config.tls_min_version

    """
     * @remarks
     * Encapsulate the request and invoke the network
     * 
     * @param action - api name
     * @param version - product version
     * @param protocol - http or https
     * @param method - e.g. GET
     * @param authType - authorization type e.g. AK
     * @param bodyType - response body type e.g. String
     * @param request - object of OpenApiRequest
     * @param runtime - which controls some details of call api, such as retry times
     * @returns the response
    """
    def do_rpcrequest(
        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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _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)

                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)

                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.'
                        )
                    credential_model = self._credential.get_credential()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.query["BearerToken"] = bearer_token
                        _request.query["SignatureType"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.query["SecurityToken"] = security_token
                        _request.query["SignatureMethod"] = 'HMAC-SHA1'
                        _request.query["SignatureVersion"] = '1.0'
                        _request.query["AccessKeyId"] = access_key_id
                        t = None
                        if not DaraCore.is_null(request.body):
                            t = request.body
                        signed_param = DaraCore.merge({}, _request.query, Utils.query(t))
                        _request.query["Signature"] = Utils.get_rpcsignature(signed_param, _request.method, access_key_secret)

                _last_request = _request
                _response = DaraCore.do_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = DaraStream.read_as_json(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = DaraStream.read_as_bytes(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = DaraStream.read_as_string(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = DaraStream.read_as_json(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = DaraStream.read_as_json(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def do_rpcrequest_async(
        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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _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)

                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)

                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.'
                        )
                    credential_model = await self._credential.get_credential_async()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.query["BearerToken"] = bearer_token
                        _request.query["SignatureType"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.query["SecurityToken"] = security_token
                        _request.query["SignatureMethod"] = 'HMAC-SHA1'
                        _request.query["SignatureVersion"] = '1.0'
                        _request.query["AccessKeyId"] = access_key_id
                        t = None
                        if not DaraCore.is_null(request.body):
                            t = request.body
                        signed_param = DaraCore.merge({}, _request.query, Utils.query(t))
                        _request.query["Signature"] = Utils.get_rpcsignature(signed_param, _request.method, access_key_secret)

                _last_request = _request
                _response = await DaraCore.async_do_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = await DaraStream.read_as_json_async(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = await DaraStream.read_as_bytes_async(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = await DaraStream.read_as_string_async(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = await DaraStream.read_as_json_async(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = await DaraStream.read_as_json_async(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    """
     * @remarks
     * Encapsulate the request and invoke the network
     * 
     * @param action - api name
     * @param version - product version
     * @param protocol - http or https
     * @param method - e.g. GET
     * @param authType - authorization type e.g. AK
     * @param pathname - pathname of every api
     * @param bodyType - response body type e.g. String
     * @param request - object of OpenApiRequest
     * @param runtime - which controls some details of call api, such as retry times
     * @returns the response
    """
    def do_roarequest(
        self,
        action: str,
        version: str,
        protocol: str,
        method: str,
        auth_type: str,
        pathname: 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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or protocol
                _request.method = method
                _request.pathname = 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.headers = DaraCore.merge({
                    'date': Utils.get_date_utcstring(),
                    'host': self._endpoint,
                    'accept': 'application/json',
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'x-acs-signature-method': 'HMAC-SHA1',
                    'x-acs-signature-version': '1.0',
                    'x-acs-version': version,
                    'x-acs-action': action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                }, global_headers, extends_headers, request.headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                if not DaraCore.is_null(request.body):
                    _request.body = DaraCore.to_json_string(request.body)
                    _request.headers["content-type"] = 'application/json; charset=utf-8'
                _request.query = DaraCore.merge({}, global_queries, extends_queries)
                if not DaraCore.is_null(request.query):
                    _request.query = DaraCore.merge({}, _request.query, request.query)
                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.'
                        )
                    credential_model = self._credential.get_credential()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        string_to_sign = Utils.get_string_to_sign(_request)
                        _request.headers["authorization"] = f'acs {access_key_id}:{Utils.get_roasignature(string_to_sign, access_key_secret)}'

                _last_request = _request
                _response = DaraCore.do_action(_request, _runtime)
                _last_response = _response
                if _response.status_code == 204:
                    return {
                        'headers': _response.headers
                    }
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = DaraStream.read_as_json(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    request_id = request_id or err.get("requestid")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = DaraStream.read_as_bytes(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = DaraStream.read_as_string(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = DaraStream.read_as_json(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = DaraStream.read_as_json(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def do_roarequest_async(
        self,
        action: str,
        version: str,
        protocol: str,
        method: str,
        auth_type: str,
        pathname: 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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or protocol
                _request.method = method
                _request.pathname = 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.headers = DaraCore.merge({
                    'date': Utils.get_date_utcstring(),
                    'host': self._endpoint,
                    'accept': 'application/json',
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'x-acs-signature-method': 'HMAC-SHA1',
                    'x-acs-signature-version': '1.0',
                    'x-acs-version': version,
                    'x-acs-action': action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                }, global_headers, extends_headers, request.headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                if not DaraCore.is_null(request.body):
                    _request.body = DaraCore.to_json_string(request.body)
                    _request.headers["content-type"] = 'application/json; charset=utf-8'
                _request.query = DaraCore.merge({}, global_queries, extends_queries)
                if not DaraCore.is_null(request.query):
                    _request.query = DaraCore.merge({}, _request.query, request.query)
                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.'
                        )
                    credential_model = await self._credential.get_credential_async()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        string_to_sign = Utils.get_string_to_sign(_request)
                        _request.headers["authorization"] = f'acs {access_key_id}:{Utils.get_roasignature(string_to_sign, access_key_secret)}'

                _last_request = _request
                _response = await DaraCore.async_do_action(_request, _runtime)
                _last_response = _response
                if _response.status_code == 204:
                    return {
                        'headers': _response.headers
                    }
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = await DaraStream.read_as_json_async(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    request_id = request_id or err.get("requestid")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = await DaraStream.read_as_bytes_async(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = await DaraStream.read_as_string_async(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = await DaraStream.read_as_json_async(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = await DaraStream.read_as_json_async(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    """
     * @remarks
     * Encapsulate the request and invoke the network with form body
     * 
     * @param action - api name
     * @param version - product version
     * @param protocol - http or https
     * @param method - e.g. GET
     * @param authType - authorization type e.g. AK
     * @param pathname - pathname of every api
     * @param bodyType - response body type e.g. String
     * @param request - object of OpenApiRequest
     * @param runtime - which controls some details of call api, such as retry times
     * @returns the response
    """
    def do_roarequest_with_form(
        self,
        action: str,
        version: str,
        protocol: str,
        method: str,
        auth_type: str,
        pathname: 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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or protocol
                _request.method = method
                _request.pathname = 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.headers = DaraCore.merge({
                    'date': Utils.get_date_utcstring(),
                    'host': self._endpoint,
                    'accept': 'application/json',
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'x-acs-signature-method': 'HMAC-SHA1',
                    'x-acs-signature-version': '1.0',
                    'x-acs-version': version,
                    'x-acs-action': action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                }, global_headers, extends_headers, request.headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                if not DaraCore.is_null(request.body):
                    m = request.body
                    _request.body = Utils.to_form(m)
                    _request.headers["content-type"] = 'application/x-www-form-urlencoded'
                _request.query = DaraCore.merge({}, global_queries, extends_queries)
                if not DaraCore.is_null(request.query):
                    _request.query = DaraCore.merge({}, _request.query, request.query)
                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.'
                        )
                    credential_model = self._credential.get_credential()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        string_to_sign = Utils.get_string_to_sign(_request)
                        _request.headers["authorization"] = f'acs {access_key_id}:{Utils.get_roasignature(string_to_sign, access_key_secret)}'

                _last_request = _request
                _response = DaraCore.do_action(_request, _runtime)
                _last_response = _response
                if _response.status_code == 204:
                    return {
                        'headers': _response.headers
                    }
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = DaraStream.read_as_json(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = DaraStream.read_as_bytes(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = DaraStream.read_as_string(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = DaraStream.read_as_json(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = DaraStream.read_as_json(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def do_roarequest_with_form_async(
        self,
        action: str,
        version: str,
        protocol: str,
        method: str,
        auth_type: str,
        pathname: 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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or protocol
                _request.method = method
                _request.pathname = 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.headers = DaraCore.merge({
                    'date': Utils.get_date_utcstring(),
                    'host': self._endpoint,
                    'accept': 'application/json',
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'x-acs-signature-method': 'HMAC-SHA1',
                    'x-acs-signature-version': '1.0',
                    'x-acs-version': version,
                    'x-acs-action': action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                }, global_headers, extends_headers, request.headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                if not DaraCore.is_null(request.body):
                    m = request.body
                    _request.body = Utils.to_form(m)
                    _request.headers["content-type"] = 'application/x-www-form-urlencoded'
                _request.query = DaraCore.merge({}, global_queries, extends_queries)
                if not DaraCore.is_null(request.query):
                    _request.query = DaraCore.merge({}, _request.query, request.query)
                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.'
                        )
                    credential_model = await self._credential.get_credential_async()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    credential_type = credential_model.type
                    if credential_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'
                    elif credential_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        string_to_sign = Utils.get_string_to_sign(_request)
                        _request.headers["authorization"] = f'acs {access_key_id}:{Utils.get_roasignature(string_to_sign, access_key_secret)}'

                _last_request = _request
                _response = await DaraCore.async_do_action(_request, _runtime)
                _last_response = _response
                if _response.status_code == 204:
                    return {
                        'headers': _response.headers
                    }
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    _res = await DaraStream.read_as_json_async(_response.body)
                    err = _res
                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif body_type == 'byte':
                    byt = await DaraStream.read_as_bytes_async(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'string':
                    _str = await DaraStream.read_as_string_async(_response.body)
                    return {
                        'body': _str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'json':
                    obj = await DaraStream.read_as_json_async(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif body_type == 'array':
                    arr = await DaraStream.read_as_json_async(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    return {
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    """
     * @remarks
     * Encapsulate the request and invoke the network
     * 
     * @param action - api name
     * @param version - product version
     * @param protocol - http or https
     * @param method - e.g. GET
     * @param authType - authorization type e.g. AK
     * @param bodyType - response body type e.g. String
     * @param request - object of OpenApiRequest
     * @param runtime - which controls some details of call api, such as retry times
     * @returns the response
    """
    def do_request(
        self,
        params: open_api_util_models.Params,
        request: open_api_util_models.OpenApiRequest,
        runtime: RuntimeOptions,
    ) -> dict:
        protocol = f'{self._protocol or params.protocol or ""}'.lower()
        is_websocket = protocol in ('ws', 'wss')
        ws_handler = None
        if is_websocket:
            websocket_sub_protocol = params.websocket_sub_protocol or ''
            if not websocket_sub_protocol:
                raise main_exceptions.ClientException(
                    code='MissingWebSocketSubProtocol',
                    message='websocketSubProtocol is required: please set it in params.websocketSubProtocol',
                )
            if websocket_sub_protocol not in (
                websocket_utils.SubProtocolAWAP,
                websocket_utils.SubProtocolGeneral,
            ):
                raise main_exceptions.ClientException(
                    code='InvalidWebSocketSubProtocol',
                    message=(
                        "websocketSubProtocol must be 'awap' or 'general', "
                        f'got: {websocket_sub_protocol}'
                    ),
                )
            if request.headers is None:
                request.headers = {}
            request.headers['sec-websocket-protocol'] = websocket_sub_protocol
            ws_handler = runtime.web_socket_handler
            if ws_handler is None:
                raise main_exceptions.ClientException(
                    code='MissingWebSocketHandler',
                    message='WebSocketHandler is required: please set it in runtime.webSocketHandler',
                )

        _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,
        }
        if is_websocket:
            _runtime.update({
                'webSocketPingInterval': get_web_socket_ping_interval(runtime),
                'webSocketPongTimeout': get_web_socket_pong_timeout(runtime),
                'webSocketEnableReconnect': get_web_socket_enable_reconnect(runtime),
                'webSocketReconnectInterval': get_web_socket_reconnect_interval(runtime),
                'webSocketMaxReconnectTimes': get_web_socket_max_reconnect_times(runtime),
                'webSocketWriteTimeout': get_web_socket_write_timeout(runtime),
                'webSocketHandshakeTimeout': get_web_socket_handshake_timeout(runtime),
                'webSocketHandler': ws_handler,
            })
        _last_request = None
        _last_response = None
        _retries_attempted = 0
        _context = RetryPolicyContext(
            retries_attempted= _retries_attempted
        )
        while DaraCore.should_retry(_runtime.get('retryOptions'), _context):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or params.protocol
                _request.method = params.method
                _request.pathname = params.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({}, global_queries, extends_queries, request.query)
                # endpoint is setted in product client
                _request.headers = DaraCore.merge({
                    'host': self._endpoint,
                    'x-acs-version': params.version,
                    'x-acs-action': params.action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                    'x-acs-date': Utils.get_timestamp(),
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'accept': 'application/json',
                }, global_headers, extends_headers, request.headers)
                if params.style == 'RPC':
                    headers = self.get_rpc_headers()
                    if not DaraCore.is_null(headers):
                        _request.headers = DaraCore.merge({}, _request.headers, headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                signature_algorithm = self._signature_algorithm or 'ACS3-HMAC-SHA256'
                hashed_request_payload = Utils.hash(DaraBytes.from_('', 'utf-8'), signature_algorithm)
                if not DaraCore.is_null(request.stream):
                    tmp = DaraStream.read_as_bytes(request.stream)
                    hashed_request_payload = Utils.hash(tmp, signature_algorithm)
                    _request.body = tmp
                    _request.headers["content-type"] = 'application/octet-stream'
                else:
                    if not DaraCore.is_null(request.body):
                        if params.req_body_type == 'byte':
                            byte_obj = bytes(request.body)
                            hashed_request_payload = Utils.hash(byte_obj, signature_algorithm)
                            _request.body = byte_obj
                        elif params.req_body_type == 'json':
                            json_obj = DaraCore.to_json_string(request.body)
                            hashed_request_payload = Utils.hash(json_obj.encode('utf-8'), signature_algorithm)
                            _request.body = json_obj
                            _request.headers["content-type"] = 'application/json; charset=utf-8'
                        else:
                            m = request.body
                            form_obj = Utils.to_form(m)
                            hashed_request_payload = Utils.hash(form_obj.encode('utf-8'), signature_algorithm)
                            _request.body = form_obj
                            _request.headers["content-type"] = 'application/x-www-form-urlencoded'


                _request.headers["x-acs-content-sha256"] = hashed_request_payload.hex()
                if params.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.'
                        )
                    credential_model = self._credential.get_credential()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    auth_type = credential_model.type
                    if auth_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        if params.style == 'RPC':
                            _request.query["SignatureType"] = 'BEARERTOKEN'
                        else:
                            _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'

                    elif auth_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        _request.headers["Authorization"] = Utils.get_authorization(_request, signature_algorithm, hashed_request_payload.hex(), access_key_id, access_key_secret)

                else:
                    if params.style == 'RPC':
                        _request.query["Format"] = 'json'

                _last_request = _request
                if is_websocket:
                    stream_handler = websocket_utils.StreamHandler(
                        ws_handler,
                        websocket_sub_protocol,
                    )
                    _runtime['webSocketHandler'] = stream_handler
                    ws_client, response_ = new_websocket_client_and_connect(_request, _runtime)
                    ws_client_obj = websocket_utils.new_websocket_client(ws_client, response_)
                    stream_handler.client = ws_client_obj
                    return {
                        'webSocketClient': ws_client_obj,
                    }

                _response = DaraCore.do_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    err = {}
                    if not DaraCore.is_null(_response.headers.get("content-type")) and _response.headers.get("content-type") == 'text/xml;charset=utf-8':
                        _str = DaraStream.read_as_string(_response.body)
                        resp_map = DaraXML.parse_xml(_str, None)
                        err = resp_map.get("Error")
                    else:
                        _res = DaraStream.read_as_json(_response.body)
                        err = _res

                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if params.body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif params.body_type == 'byte':
                    byt = DaraStream.read_as_bytes(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'string':
                    resp_str = DaraStream.read_as_string(_response.body)
                    return {
                        'body': resp_str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'json':
                    obj = DaraStream.read_as_json(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'array':
                    arr = DaraStream.read_as_json(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    anything = DaraStream.read_as_string(_response.body)
                    return {
                        'body': anything,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def do_request_async(
        self,
        params: open_api_util_models.Params,
        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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or params.protocol
                _request.method = params.method
                _request.pathname = params.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({}, global_queries, extends_queries, request.query)
                # endpoint is setted in product client
                _request.headers = DaraCore.merge({
                    'host': self._endpoint,
                    'x-acs-version': params.version,
                    'x-acs-action': params.action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                    'x-acs-date': Utils.get_timestamp(),
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'accept': 'application/json',
                }, global_headers, extends_headers, request.headers)
                if params.style == 'RPC':
                    headers = self.get_rpc_headers()
                    if not DaraCore.is_null(headers):
                        _request.headers = DaraCore.merge({}, _request.headers, headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                signature_algorithm = self._signature_algorithm or 'ACS3-HMAC-SHA256'
                hashed_request_payload = Utils.hash(DaraBytes.from_('', 'utf-8'), signature_algorithm)
                if not DaraCore.is_null(request.stream):
                    tmp = await DaraStream.read_as_bytes_async(request.stream)
                    hashed_request_payload = Utils.hash(tmp, signature_algorithm)
                    _request.body = tmp
                    _request.headers["content-type"] = 'application/octet-stream'
                else:
                    if not DaraCore.is_null(request.body):
                        if params.req_body_type == 'byte':
                            byte_obj = bytes(request.body)
                            hashed_request_payload = Utils.hash(byte_obj, signature_algorithm)
                            _request.body = byte_obj
                        elif params.req_body_type == 'json':
                            json_obj = DaraCore.to_json_string(request.body)
                            hashed_request_payload = Utils.hash(json_obj.encode('utf-8'), signature_algorithm)
                            _request.body = json_obj
                            _request.headers["content-type"] = 'application/json; charset=utf-8'
                        else:
                            m = request.body
                            form_obj = Utils.to_form(m)
                            hashed_request_payload = Utils.hash(form_obj.encode('utf-8'), signature_algorithm)
                            _request.body = form_obj
                            _request.headers["content-type"] = 'application/x-www-form-urlencoded'


                _request.headers["x-acs-content-sha256"] = hashed_request_payload.hex()
                if params.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.'
                        )
                    credential_model = await self._credential.get_credential_async()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    auth_type = credential_model.type
                    if auth_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                        if params.style == 'RPC':
                            _request.query["SignatureType"] = 'BEARERTOKEN'
                        else:
                            _request.headers["x-acs-signature-type"] = 'BEARERTOKEN'

                    elif auth_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        _request.headers["Authorization"] = Utils.get_authorization(_request, signature_algorithm, hashed_request_payload.hex(), access_key_id, access_key_secret)

                else:
                    if params.style == 'RPC':
                        _request.query["Format"] = 'json'

                _last_request = _request
                _response = await DaraCore.async_do_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    err = {}
                    if not DaraCore.is_null(_response.headers.get("content-type")) and _response.headers.get("content-type") == 'text/xml;charset=utf-8':
                        _str = await DaraStream.read_as_string_async(_response.body)
                        resp_map = DaraXML.parse_xml(_str, None)
                        err = resp_map.get("Error")
                    else:
                        _res = await DaraStream.read_as_json_async(_response.body)
                        err = _res

                    request_id = err.get("RequestId") or err.get("requestId")
                    code = err.get("Code") or err.get("code")
                    retry_after = Utils.get_throttling_time_left(_response.headers)
                    if not DaraCore.is_null(retry_after):
                        raise main_exceptions.ThrottlingException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            retry_after = retry_after,
                            data = err,
                            request_id = f'{request_id}'
                        )
                    elif (_response.status_code >= 400) and (_response.status_code < 500):
                        raise main_exceptions.ClientException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            access_denied_detail = self.get_access_denied_detail(err),
                            request_id = f'{request_id}'
                        )
                    else:
                        raise main_exceptions.ServerException(
                            status_code = _response.status_code,
                            code = f'{code}',
                            message = f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {request_id}',
                            detail = f'{err.get("Detail") or err.get("detail")}',
                            description = f'{err.get("Description") or err.get("description")}',
                            data = err,
                            request_id = f'{request_id}'
                        )

                if params.body_type == 'binary':
                    resp = {
                        'body': _response.body,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                    return resp
                elif params.body_type == 'byte':
                    byt = await DaraStream.read_as_bytes_async(_response.body)
                    return {
                        'body': byt,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'string':
                    resp_str = await DaraStream.read_as_string_async(_response.body)
                    return {
                        'body': resp_str,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'json':
                    obj = await DaraStream.read_as_json_async(_response.body)
                    res = obj
                    return {
                        'body': res,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                elif params.body_type == 'array':
                    arr = await DaraStream.read_as_json_async(_response.body)
                    return {
                        'body': arr,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }
                else:
                    anything = await DaraStream.read_as_string_async(_response.body)
                    return {
                        'body': anything,
                        'headers': _response.headers,
                        'statusCode': _response.status_code
                    }

            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    """
     * @remarks
     * Encapsulate the request and invoke the network
     * 
     * @param action - api name
     * @param version - product version
     * @param protocol - http or https
     * @param method - e.g. GET
     * @param authType - authorization type e.g. AK
     * @param bodyType - response body type e.g. String
     * @param request - object of OpenApiRequest
     * @param runtime - which controls some details of call api, such as retry times
     * @returns the response
    """
    def execute(
        self,
        params: open_api_util_models.Params,
        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,
            'disableHttp2': bool(self._disable_http_2 or False),
        }
        _last_request = None
        _last_response = None
        _retries_attempted = 0
        _context = RetryPolicyContext(
            retries_attempted= _retries_attempted
        )
        while DaraCore.should_retry(_runtime.get('retryOptions'), _context):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                # spi = new Gateway();//Gateway implements SPI，这一步在产品 SDK 中实例化
                headers = self.get_rpc_headers()
                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_context = spi_models.InterceptorContextRequest(
                    headers = DaraCore.merge({}, global_headers, extends_headers, request.headers, headers),
                    query = DaraCore.merge({}, global_queries, extends_queries, request.query),
                    body = request.body,
                    stream = request.stream,
                    host_map = request.host_map,
                    pathname = params.pathname,
                    product_id = self._product_id,
                    action = params.action,
                    version = params.version,
                    protocol = self._protocol or params.protocol,
                    method = self._method or params.method,
                    auth_type = params.auth_type,
                    body_type = params.body_type,
                    req_body_type = params.req_body_type,
                    style = params.style,
                    credential = self._credential,
                    signature_version = self._signature_version,
                    signature_algorithm = self._signature_algorithm,
                    user_agent = Utils.get_user_agent(self._user_agent)
                )
                configuration_context = spi_models.InterceptorContextConfiguration(
                    region_id = self._region_id,
                    endpoint = request.endpoint_override or self._endpoint,
                    endpoint_rule = self._endpoint_rule,
                    endpoint_map = self._endpoint_map,
                    endpoint_type = self._endpoint_type,
                    network = self._network,
                    suffix = self._suffix
                )
                interceptor_context = spi_models.InterceptorContext(
                    request = request_context,
                    configuration = configuration_context
                )
                attribute_map = spi_models.AttributeMap()
                if not DaraCore.is_null(self._attribute_map):
                    attribute_map = self._attribute_map
                # 1. spi.modifyConfiguration(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                self._spi.modify_configuration(interceptor_context, attribute_map)
                # 2. spi.modifyRequest(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                self._spi.modify_request(interceptor_context, attribute_map)
                _request.protocol = interceptor_context.request.protocol
                _request.method = interceptor_context.request.method
                _request.pathname = interceptor_context.request.pathname
                _request.query = interceptor_context.request.query
                _request.body = interceptor_context.request.stream
                _request.headers = interceptor_context.request.headers
                _last_request = _request
                _response = DaraCore.do_action(_request, _runtime)
                _last_response = _response
                response_context = spi_models.InterceptorContextResponse(
                    status_code = _response.status_code,
                    headers = _response.headers,
                    body = _response.body
                )
                interceptor_context.response = response_context
                # 3. spi.modifyResponse(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                self._spi.modify_response(interceptor_context, attribute_map)
                resp = {
                    'headers': interceptor_context.response.headers,
                    'statusCode': interceptor_context.response.status_code,
                    'body': interceptor_context.response.deserialized_body
                }
                return resp
            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def execute_async(
        self,
        params: open_api_util_models.Params,
        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,
            'disableHttp2': bool(self._disable_http_2 or False),
        }
        _last_request = None
        _last_response = None
        _retries_attempted = 0
        _context = RetryPolicyContext(
            retries_attempted= _retries_attempted
        )
        while DaraCore.should_retry(_runtime.get('retryOptions'), _context):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                # spi = new Gateway();//Gateway implements SPI，这一步在产品 SDK 中实例化
                headers = self.get_rpc_headers()
                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_context = spi_models.InterceptorContextRequest(
                    headers = DaraCore.merge({}, global_headers, extends_headers, request.headers, headers),
                    query = DaraCore.merge({}, global_queries, extends_queries, request.query),
                    body = request.body,
                    stream = request.stream,
                    host_map = request.host_map,
                    pathname = params.pathname,
                    product_id = self._product_id,
                    action = params.action,
                    version = params.version,
                    protocol = self._protocol or params.protocol,
                    method = self._method or params.method,
                    auth_type = params.auth_type,
                    body_type = params.body_type,
                    req_body_type = params.req_body_type,
                    style = params.style,
                    credential = self._credential,
                    signature_version = self._signature_version,
                    signature_algorithm = self._signature_algorithm,
                    user_agent = Utils.get_user_agent(self._user_agent)
                )
                configuration_context = spi_models.InterceptorContextConfiguration(
                    region_id = self._region_id,
                    endpoint = request.endpoint_override or self._endpoint,
                    endpoint_rule = self._endpoint_rule,
                    endpoint_map = self._endpoint_map,
                    endpoint_type = self._endpoint_type,
                    network = self._network,
                    suffix = self._suffix
                )
                interceptor_context = spi_models.InterceptorContext(
                    request = request_context,
                    configuration = configuration_context
                )
                attribute_map = spi_models.AttributeMap()
                if not DaraCore.is_null(self._attribute_map):
                    attribute_map = self._attribute_map
                # 1. spi.modifyConfiguration(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                await self._spi.modify_configuration_async(interceptor_context, attribute_map)
                # 2. spi.modifyRequest(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                await self._spi.modify_request_async(interceptor_context, attribute_map)
                _request.protocol = interceptor_context.request.protocol
                _request.method = interceptor_context.request.method
                _request.pathname = interceptor_context.request.pathname
                _request.query = interceptor_context.request.query
                _request.body = interceptor_context.request.stream
                _request.headers = interceptor_context.request.headers
                _last_request = _request
                _response = await DaraCore.async_do_action(_request, _runtime)
                _last_response = _response
                response_context = spi_models.InterceptorContextResponse(
                    status_code = _response.status_code,
                    headers = _response.headers,
                    body = _response.body
                )
                interceptor_context.response = response_context
                # 3. spi.modifyResponse(context: SPI.InterceptorContext, attributeMap: SPI.AttributeMap);
                await self._spi.modify_response_async(interceptor_context, attribute_map)
                resp = {
                    'headers': interceptor_context.response.headers,
                    'statusCode': interceptor_context.response.status_code,
                    'body': interceptor_context.response.deserialized_body
                }
                return resp
            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    def call_sseapi(
        self,
        params: open_api_util_models.Params,
        request: open_api_util_models.OpenApiRequest,
        runtime: RuntimeOptions,
    ) -> Generator[main_models.SSEResponse, None, None]:
        _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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or params.protocol
                _request.method = params.method
                _request.pathname = params.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({}, global_queries, extends_queries, request.query)
                # endpoint is setted in product client
                _request.headers = DaraCore.merge({
                    'host': self._endpoint,
                    'x-acs-version': params.version,
                    'x-acs-action': params.action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                    'x-acs-date': Utils.get_timestamp(),
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'accept': 'application/json',
                }, extends_headers, global_headers, request.headers)
                if params.style == 'RPC':
                    headers = self.get_rpc_headers()
                    if not DaraCore.is_null(headers):
                        _request.headers = DaraCore.merge({}, _request.headers, headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                signature_algorithm = self._signature_algorithm or 'ACS3-HMAC-SHA256'
                hashed_request_payload = Utils.hash(DaraBytes.from_('', 'utf-8'), signature_algorithm)
                if not DaraCore.is_null(request.stream):
                    tmp = DaraStream.read_as_bytes(request.stream)
                    hashed_request_payload = Utils.hash(tmp, signature_algorithm)
                    _request.body = tmp
                    _request.headers["content-type"] = 'application/octet-stream'
                else:
                    if not DaraCore.is_null(request.body):
                        if params.req_body_type == 'byte':
                            byte_obj = bytes(request.body)
                            hashed_request_payload = Utils.hash(byte_obj, signature_algorithm)
                            _request.body = byte_obj
                        elif params.req_body_type == 'json':
                            json_obj = DaraCore.to_json_string(request.body)
                            hashed_request_payload = Utils.hash(json_obj.encode('utf-8'), signature_algorithm)
                            _request.body = json_obj
                            _request.headers["content-type"] = 'application/json; charset=utf-8'
                        else:
                            m = request.body
                            form_obj = Utils.to_form(m)
                            hashed_request_payload = Utils.hash(form_obj.encode('utf-8'), signature_algorithm)
                            _request.body = form_obj
                            _request.headers["content-type"] = 'application/x-www-form-urlencoded'


                _request.headers["x-acs-content-sha256"] = hashed_request_payload.hex()
                if params.auth_type != 'Anonymous':
                    credential_model = self._credential.get_credential()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    auth_type = credential_model.type
                    if auth_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                    elif auth_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        _request.headers["Authorization"] = Utils.get_authorization(_request, signature_algorithm, hashed_request_payload.hex(), access_key_id, access_key_secret)

                _last_request = _request
                _response = DaraCore.do_sse_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    err = {}
                    if not DaraCore.is_null(_response.headers.get("content-type")) and _response.headers.get("content-type") == 'text/xml;charset=utf-8':
                        _str = DaraStream.read_as_string(_response.body)
                        resp_map = DaraXML.parse_xml(_str, None)
                        err = resp_map.get("Error")
                    else:
                        _res = DaraStream.read_as_json(_response.body)
                        err = _res

                    err["statusCode"] = _response.status_code
                    raise DaraException({
                        'code': f'{err.get("Code") or err.get("code")}',
                        'message': f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {err.get("RequestId") or err.get("requestId")}',
                        'data': err,
                        'detail': f'{err.get("Detail") or err.get("detail")}',
                        'description': f'{err.get("Description") or err.get("description")}',
                        'accessDeniedDetail': err.get("AccessDeniedDetail") or err.get("accessDeniedDetail")
                    })
                events = DaraStream.read_as_sse(_response.body)
                for event in events:
                    yield  main_models.SSEResponse(
                        status_code = _response.status_code,
                        headers = _response.headers,
                        event = event
                    )
                return
            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    async def call_sseapi_async(
        self,
        params: open_api_util_models.Params,
        request: open_api_util_models.OpenApiRequest,
        runtime: RuntimeOptions,
    ) -> AsyncGenerator[main_models.SSEResponse, None, None]:
        _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):
            _backoff_time = 0
            if _retries_attempted > 0:
                _backoff_time = DaraCore.get_backoff_time(_runtime.get('retryOptions'), _context)
                if _backoff_time > 0:
                    DaraCore.sleep(_backoff_time)
            _retry_attempts = _retries_attempted
            _retries_attempted = _retries_attempted + 1
            try:
                _request = DaraRequest()
                _request.protocol = self._protocol or params.protocol
                _request.method = params.method
                _request.pathname = params.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({}, global_queries, extends_queries, request.query)
                # endpoint is setted in product client
                _request.headers = DaraCore.merge({
                    'host': self._endpoint,
                    'x-acs-version': params.version,
                    'x-acs-action': params.action,
                    'user-agent': Utils.get_user_agent(self._user_agent),
                    'x-acs-date': Utils.get_timestamp(),
                    'x-acs-signature-nonce': Utils.get_nonce(),
                    'accept': 'application/json',
                }, extends_headers, global_headers, request.headers)
                if params.style == 'RPC':
                    headers = self.get_rpc_headers()
                    if not DaraCore.is_null(headers):
                        _request.headers = DaraCore.merge({}, _request.headers, headers)
                Utils.apply_retry_headers(_request.headers, _retry_attempts, _backoff_time)
                signature_algorithm = self._signature_algorithm or 'ACS3-HMAC-SHA256'
                hashed_request_payload = Utils.hash(DaraBytes.from_('', 'utf-8'), signature_algorithm)
                if not DaraCore.is_null(request.stream):
                    tmp = await DaraStream.read_as_bytes_async(request.stream)
                    hashed_request_payload = Utils.hash(tmp, signature_algorithm)
                    _request.body = tmp
                    _request.headers["content-type"] = 'application/octet-stream'
                else:
                    if not DaraCore.is_null(request.body):
                        if params.req_body_type == 'byte':
                            byte_obj = bytes(request.body)
                            hashed_request_payload = Utils.hash(byte_obj, signature_algorithm)
                            _request.body = byte_obj
                        elif params.req_body_type == 'json':
                            json_obj = DaraCore.to_json_string(request.body)
                            hashed_request_payload = Utils.hash(json_obj.encode('utf-8'), signature_algorithm)
                            _request.body = json_obj
                            _request.headers["content-type"] = 'application/json; charset=utf-8'
                        else:
                            m = request.body
                            form_obj = Utils.to_form(m)
                            hashed_request_payload = Utils.hash(form_obj.encode('utf-8'), signature_algorithm)
                            _request.body = form_obj
                            _request.headers["content-type"] = 'application/x-www-form-urlencoded'


                _request.headers["x-acs-content-sha256"] = hashed_request_payload.hex()
                if params.auth_type != 'Anonymous':
                    credential_model = await self._credential.get_credential_async()
                    if not DaraCore.is_null(credential_model.provider_name):
                        _request.headers["x-acs-credentials-provider"] = credential_model.provider_name
                    auth_type = credential_model.type
                    if auth_type == 'bearer':
                        bearer_token = credential_model.bearer_token
                        _request.headers["x-acs-bearer-token"] = bearer_token
                    elif auth_type == 'id_token':
                        id_token = credential_model.security_token
                        _request.headers["x-acs-zero-trust-idtoken"] = id_token
                    else:
                        access_key_id = credential_model.access_key_id
                        access_key_secret = credential_model.access_key_secret
                        security_token = credential_model.security_token
                        if not DaraCore.is_null(security_token) and security_token != '':
                            _request.headers["x-acs-accesskey-id"] = access_key_id
                            _request.headers["x-acs-security-token"] = security_token
                        _request.headers["Authorization"] = Utils.get_authorization(_request, signature_algorithm, hashed_request_payload.hex(), access_key_id, access_key_secret)

                _last_request = _request
                _response = await DaraCore.async_do_sse_action(_request, _runtime)
                _last_response = _response
                if (_response.status_code >= 400) and (_response.status_code < 600):
                    err = {}
                    if not DaraCore.is_null(_response.headers.get("content-type")) and _response.headers.get("content-type") == 'text/xml;charset=utf-8':
                        _str = await DaraStream.read_as_string_async(_response.body)
                        resp_map = DaraXML.parse_xml(_str, None)
                        err = resp_map.get("Error")
                    else:
                        _res = await DaraStream.read_as_json_async(_response.body)
                        err = _res

                    err["statusCode"] = _response.status_code
                    raise DaraException({
                        'code': f'{err.get("Code") or err.get("code")}',
                        'message': f'code: {_response.status_code}, {err.get("Message") or err.get("message")} request id: {err.get("RequestId") or err.get("requestId")}',
                        'data': err,
                        'detail': f'{err.get("Detail") or err.get("detail")}',
                        'description': f'{err.get("Description") or err.get("description")}',
                        'accessDeniedDetail': err.get("AccessDeniedDetail") or err.get("accessDeniedDetail")
                    })
                events = DaraStream.read_as_sse_async(_response.body)
                async for event in events:
                    yield  main_models.SSEResponse(
                        status_code = _response.status_code,
                        headers = _response.headers,
                        event = event
                    )
                return
            except Exception as e:
                _context = RetryPolicyContext(
                    retries_attempted= _retries_attempted,
                    http_request = _last_request,
                    http_response = _last_response,
                    exception = e
                )
                continue
        raise UnretryableException(_context)

    def call_api(
        self,
        params: open_api_util_models.Params,
        request: open_api_util_models.OpenApiRequest,
        runtime: RuntimeOptions,
    ) -> dict:
        if DaraCore.is_null(params):
            raise main_exceptions.ClientException(
                code = 'ParameterMissing',
                message = '\'params\' can not be unset'
            )
        if DaraCore.is_null(self._signature_version) or self._signature_version != 'v4':
            if DaraCore.is_null(self._signature_algorithm) or self._signature_algorithm != 'v2':
                return self.do_request(params, request, runtime)
            elif (params.style == 'ROA') and (params.req_body_type == 'json'):
                return self.do_roarequest(params.action, params.version, params.protocol, params.method, params.auth_type, params.pathname, params.body_type, request, runtime)
            elif params.style == 'ROA':
                return self.do_roarequest_with_form(params.action, params.version, params.protocol, params.method, params.auth_type, params.pathname, params.body_type, request, runtime)
            else:
                return self.do_rpcrequest(params.action, params.version, params.protocol, params.method, params.auth_type, params.body_type, request, runtime)

        else:
            return self.execute(params, request, runtime)


    async def call_api_async(
        self,
        params: open_api_util_models.Params,
        request: open_api_util_models.OpenApiRequest,
        runtime: RuntimeOptions,
    ) -> dict:
        if DaraCore.is_null(params):
            raise main_exceptions.ClientException(
                code = 'ParameterMissing',
                message = '\'params\' can not be unset'
            )
        if DaraCore.is_null(self._signature_version) or self._signature_version != 'v4':
            if DaraCore.is_null(self._signature_algorithm) or self._signature_algorithm != 'v2':
                return await self.do_request_async(params, request, runtime)
            elif (params.style == 'ROA') and (params.req_body_type == 'json'):
                return await self.do_roarequest_async(params.action, params.version, params.protocol, params.method, params.auth_type, params.pathname, params.body_type, request, runtime)
            elif params.style == 'ROA':
                return await self.do_roarequest_with_form_async(params.action, params.version, params.protocol, params.method, params.auth_type, params.pathname, params.body_type, request, runtime)
            else:
                return await self.do_rpcrequest_async(params.action, params.version, params.protocol, params.method, params.auth_type, params.body_type, request, runtime)

        else:
            return await self.execute_async(params, request, runtime)


    def get_access_key_id(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        access_key_id = self._credential.get_access_key_id()
        return access_key_id

    async def get_access_key_id_async(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        access_key_id = await self._credential.get_access_key_id_async()
        return access_key_id

    def get_access_key_secret(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        secret = self._credential.get_access_key_secret()
        return secret

    async def get_access_key_secret_async(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        secret = await self._credential.get_access_key_secret_async()
        return secret

    def get_security_token(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        token = self._credential.get_security_token()
        return token

    async def get_security_token_async(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        token = await self._credential.get_security_token_async()
        return token

    def get_bearer_token(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        token = self._credential.get_bearer_token()
        return token

    async def get_bearer_token_async(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        token = self._credential.get_bearer_token()
        return token

    def get_type(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        auth_type = self._credential.get_type()
        return auth_type

    async def get_type_async(self) -> str:
        if DaraCore.is_null(self._credential):
            return ''
        auth_type = self._credential.get_type()
        return auth_type

    def check_config(
        self,
        config: open_api_util_models.Config,
    ) -> None:
        if DaraCore.is_null(self._endpoint_rule) and DaraCore.is_null(config.endpoint):
            raise main_exceptions.ClientException(
                code = 'ParameterMissing',
                message = '\'config.endpoint\' can not be empty'
            )

    def set_gateway_client(
        self,
        spi: SPIClient,
    ) -> None:
        self._spi = spi

    def set_rpc_headers(
        self,
        headers: Dict[str, str],
    ) -> None:
        self._headers = headers

    def get_rpc_headers(self) -> Dict[str, str]:
        headers = self._headers
        self._headers = None
        return headers

    def default_any(
        input_value: Any,
        default_value: Any,
    ) -> Any:
        if DaraCore.is_null(input_value):
            return default_value
        return input_value

    def get_access_denied_detail(
        self,
        err: Dict[str, Any],
    ) -> Dict[str, Any]:
        access_denied_detail = None
        if not DaraCore.is_null(err.get("AccessDeniedDetail")):
            detail_1 = err.get("AccessDeniedDetail")
            access_denied_detail = detail_1
        elif not DaraCore.is_null(err.get("accessDeniedDetail")):
            detail_2 = err.get("accessDeniedDetail")
            access_denied_detail = detail_2
        return access_denied_detail