alibabacloud-gateway-oss/python/alibabacloud_gateway_oss/client.py

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
from alibabacloud_darabonba_encode_util.encoder import Encoder
from alibabacloud_darabonba_signature_util.signer import Signer
from Tea.exceptions import TeaException
from typing import List, Dict
from Tea.core import TeaCore

from alibabacloud_gateway_spi.client import Client as SPIClient
from alibabacloud_gateway_spi import models as spi_models
from alibabacloud_tea_util.client import Client as UtilClient
from alibabacloud_darabonba_map.client import Client as MapClient
from alibabacloud_darabonba_string.client import Client as StringClient
from alibabacloud_tea_xml.client import Client as XMLClient
from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
from alibabacloud_oss_util.client import Client as OSSUtilClient
from alibabacloud_darabonba_array.client import Client as ArrayClient
from alibabacloud_gateway_oss_util.client import Client as OSS_UtilClient


class Client(SPIClient):
    _default_signed_params: List[str] = None
    _except_signed_params: List[str] = None

    def __init__(self):
        super().__init__()
        self._default_signed_params = [
            'response-content-type',
            'response-content-language',
            'response-cache-control',
            'logging',
            'response-content-encoding',
            'acl',
            'uploadId',
            'uploads',
            'partNumber',
            'group',
            'link',
            'delete',
            'website',
            'location',
            'objectInfo',
            'objectMeta',
            'response-expires',
            'response-content-disposition',
            'cors',
            'lifecycle',
            'restore',
            'qos',
            'referer',
            'stat',
            'bucketInfo',
            'append',
            'position',
            'security-token',
            'live',
            'comp',
            'status',
            'vod',
            'startTime',
            'endTime',
            'x-oss-process',
            'symlink',
            'callback',
            'callback-var',
            'tagging',
            'encryption',
            'versions',
            'versioning',
            'versionId',
            'policy',
            'requestPayment',
            'x-oss-traffic-limit',
            'qosInfo',
            'asyncFetch',
            'x-oss-request-payer',
            'sequential',
            'inventory',
            'inventoryId',
            'continuation-token',
            'callback',
            'callback-var',
            'worm',
            'wormId',
            'wormExtend',
            'replication',
            'replicationLocation',
            'replicationProgress',
            'transferAcceleration',
            'cname',
            'metaQuery',
            'x-oss-ac-source-ip',
            'x-oss-ac-subnet-mask',
            'x-oss-ac-vpc-id',
            'x-oss-ac-forward-allow',
            'resourceGroup',
            'style',
            'styleName',
            'x-oss-async-process',
            'rtc',
            'accessPoint',
            'accessPointPolicy',
            'httpsConfig',
            'regionsV2',
            'publicAccessBlock',
            'policyStatus',
            'redundancyTransition',
            'redundancyType',
            'redundancyProgress',
            'dataAccelerator',
            'verbose',
            'accessPointForObjectProcess',
            'accessPointConfigForObjectProcess',
            'accessPointPolicyForObjectProcess',
            'bucketArchiveDirectRead',
            'responseHeader',
            'userDefinedLogFieldsConfig',
            'reservedcapacity',
            'requesterQosInfo',
            'qosRequester',
            'resourcePool',
            'resourcePoolInfo',
            'resourcePoolBuckets',
            'processConfiguration',
            'img',
            'asyncFetch',
            'virtualBucket',
            'copy',
            'userRegion',
            'partSize',
            'chunkSize',
            'partUploadId',
            'chunkNumber',
            'userRegion',
            'regionList',
            'eventnotification',
            'cacheConfiguration',
            'dfs',
            'dfsadmin',
            'dfssecurity'
        ]
        self._except_signed_params = [
            'list-type',
            'regions'
        ]

    def modify_configuration(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        config = context.configuration
        config.endpoint = self.get_endpoint(config.region_id, config.network, config.endpoint)

    async def modify_configuration_async(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        config = context.configuration
        config.endpoint = await self.get_endpoint_async(config.region_id, config.network, config.endpoint)

    def modify_request(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        request = context.request
        host_map = {}
        if not UtilClient.is_unset(request.host_map):
            host_map = request.host_map
        bucket_name = host_map.get('bucket')
        if UtilClient.is_unset(bucket_name):
            bucket_name = ''
        if not UtilClient.is_unset(request.headers.get('x-oss-meta-*')):
            tmp = UtilClient.parse_json(request.headers.get('x-oss-meta-*'))
            map_data = UtilClient.assert_as_map(tmp)
            meta_data = UtilClient.stringify_map_value(map_data)
            meta_key_set = MapClient.key_set(meta_data)
            request.headers['x-oss-meta-*'] = None
            for key in meta_key_set:
                new_key = f'x-oss-meta-{key}'
                request.headers[new_key] = meta_data.get(key)
        config = context.configuration
        region_id = config.region_id
        if UtilClient.is_unset(region_id) or UtilClient.empty(region_id):
            region_id = self.get_region_id_from_endpoint(config.endpoint)
        credential = request.credential
        access_key_id = credential.get_access_key_id()
        access_key_secret = credential.get_access_key_secret()
        security_token = credential.get_security_token()
        if not UtilClient.empty(security_token):
            request.headers['x-oss-security-token'] = security_token
        if not UtilClient.is_unset(request.body):
            if StringClient.equals(request.req_body_type, 'xml'):
                req_body_map = UtilClient.assert_as_map(request.body)
                xml_str = OSS_UtilClient.to_xml(req_body_map)
                request.stream = xml_str
                request.headers['content-type'] = 'application/xml'
                request.headers['content-md5'] = Encoder.base_64encode_to_string(Signer.md5sign(xml_str))
            elif StringClient.equals(request.req_body_type, 'json'):
                req_body_str = UtilClient.to_jsonstring(request.body)
                request.stream = req_body_str
                request.headers['content-type'] = 'application/json; charset=utf-8'
            elif StringClient.equals(request.req_body_type, 'formData'):
                req_body_form = UtilClient.assert_as_map(request.body)
                request.stream = OpenApiUtilClient.to_form(req_body_form)
                request.headers['content-type'] = 'application/x-www-form-urlencoded'
            elif StringClient.equals(request.req_body_type, 'binary'):
                attribute_map.key = {
                    'crc': '',
                    'md5': ''
                }
                request.stream = OSSUtilClient.inject(request.stream, attribute_map.key)
                request.headers['content-type'] = 'application/octet-stream'
        host = self.get_host(config.endpoint_type, bucket_name, config.endpoint, context)
        request.headers = TeaCore.merge({
            'host': host,
            'date': UtilClient.get_date_utcstring(),
            'user-agent': request.user_agent
        }, request.headers)
        origin_path = request.pathname
        origin_query = request.query
        if not UtilClient.empty(origin_path):
            path_and_queries = StringClient.split(origin_path, f'?', 2)
            request.pathname = path_and_queries[0]
            if UtilClient.equal_number(ArrayClient.size(path_and_queries), 2):
                path_queries = StringClient.split(path_and_queries[1], '&', None)
                for sub in path_queries:
                    item = StringClient.split(sub, '=', None)
                    query_key = item[0]
                    query_value = ''
                    if UtilClient.equal_number(ArrayClient.size(item), 2):
                        query_value = item[1]
                    if UtilClient.empty(origin_query.get(query_key)):
                        request.query[query_key] = query_value
        signature_version = UtilClient.default_string(request.signature_version, 'v4')
        request.headers['authorization'] = self.get_authorization(signature_version, bucket_name, request.pathname, request.method, request.query, request.headers, access_key_id, access_key_secret, region_id)

    async def modify_request_async(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        request = context.request
        host_map = {}
        if not UtilClient.is_unset(request.host_map):
            host_map = request.host_map
        bucket_name = host_map.get('bucket')
        if UtilClient.is_unset(bucket_name):
            bucket_name = ''
        if not UtilClient.is_unset(request.headers.get('x-oss-meta-*')):
            tmp = UtilClient.parse_json(request.headers.get('x-oss-meta-*'))
            map_data = UtilClient.assert_as_map(tmp)
            meta_data = UtilClient.stringify_map_value(map_data)
            meta_key_set = MapClient.key_set(meta_data)
            request.headers['x-oss-meta-*'] = None
            for key in meta_key_set:
                new_key = f'x-oss-meta-{key}'
                request.headers[new_key] = meta_data.get(key)
        config = context.configuration
        region_id = config.region_id
        if UtilClient.is_unset(region_id) or UtilClient.empty(region_id):
            region_id = await self.get_region_id_from_endpoint_async(config.endpoint)
        credential = request.credential
        access_key_id = await credential.get_access_key_id_async()
        access_key_secret = await credential.get_access_key_secret_async()
        security_token = await credential.get_security_token_async()
        if not UtilClient.empty(security_token):
            request.headers['x-oss-security-token'] = security_token
        if not UtilClient.is_unset(request.body):
            if StringClient.equals(request.req_body_type, 'xml'):
                req_body_map = UtilClient.assert_as_map(request.body)
                xml_str = OSS_UtilClient.to_xml(req_body_map)
                request.stream = xml_str
                request.headers['content-type'] = 'application/xml'
                request.headers['content-md5'] = Encoder.base_64encode_to_string(Signer.md5sign(xml_str))
            elif StringClient.equals(request.req_body_type, 'json'):
                req_body_str = UtilClient.to_jsonstring(request.body)
                request.stream = req_body_str
                request.headers['content-type'] = 'application/json; charset=utf-8'
            elif StringClient.equals(request.req_body_type, 'formData'):
                req_body_form = UtilClient.assert_as_map(request.body)
                request.stream = OpenApiUtilClient.to_form(req_body_form)
                request.headers['content-type'] = 'application/x-www-form-urlencoded'
            elif StringClient.equals(request.req_body_type, 'binary'):
                attribute_map.key = {
                    'crc': '',
                    'md5': ''
                }
                request.stream = OSSUtilClient.inject(request.stream, attribute_map.key)
                request.headers['content-type'] = 'application/octet-stream'
        host = await self.get_host_async(config.endpoint_type, bucket_name, config.endpoint, context)
        request.headers = TeaCore.merge({
            'host': host,
            'date': UtilClient.get_date_utcstring(),
            'user-agent': request.user_agent
        }, request.headers)
        origin_path = request.pathname
        origin_query = request.query
        if not UtilClient.empty(origin_path):
            path_and_queries = StringClient.split(origin_path, f'?', 2)
            request.pathname = path_and_queries[0]
            if UtilClient.equal_number(ArrayClient.size(path_and_queries), 2):
                path_queries = StringClient.split(path_and_queries[1], '&', None)
                for sub in path_queries:
                    item = StringClient.split(sub, '=', None)
                    query_key = item[0]
                    query_value = ''
                    if UtilClient.equal_number(ArrayClient.size(item), 2):
                        query_value = item[1]
                    if UtilClient.empty(origin_query.get(query_key)):
                        request.query[query_key] = query_value
        signature_version = UtilClient.default_string(request.signature_version, 'v4')
        request.headers['authorization'] = await self.get_authorization_async(signature_version, bucket_name, request.pathname, request.method, request.query, request.headers, access_key_id, access_key_secret, region_id)

    def modify_response(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        request = context.request
        response = context.response
        body_str = None
        if UtilClient.is_4xx(response.status_code) or UtilClient.is_5xx(response.status_code):
            body_str = UtilClient.read_as_string(response.body)
            if not UtilClient.empty(body_str):
                resp_map = XMLClient.parse_xml(body_str, None)
                err = UtilClient.assert_as_map(resp_map.get('Error'))
                raise TeaException({
                    'code': err.get('Code'),
                    'message': err.get('Message'),
                    'data': {
                        'statusCode': response.status_code,
                        'requestId': err.get('RequestId'),
                        'ecCode': err.get('EC'),
                        'Recommend': err.get('RecommendDoc'),
                        'hostId': err.get('HostId'),
                        'AccessDeniedDetail': err.get('AccessDeniedDetail')
                    }
                })
            else:
                headers = response.headers
                request_id = headers.get('x-oss-request-id')
                ec_code = headers.get('x-oss-ec-code')
                raise TeaException({
                    'code': response.status_code,
                    'message': None,
                    'data': {
                        'statusCode': response.status_code,
                        'requestId': f'{request_id}',
                        'ecCode': ec_code
                    }
                })
        ctx = attribute_map.key
        if not UtilClient.is_unset(ctx):
            if not UtilClient.is_unset(ctx.get('crc')) and not UtilClient.is_unset(response.headers.get('x-oss-hash-crc64ecma')) and not StringClient.equals(ctx.get('crc'), response.headers.get('x-oss-hash-crc64ecma')):
                raise TeaException({
                    'code': 'CrcNotMatched',
                    'data': {
                        'clientCrc': ctx.get('crc'),
                        'serverCrc': response.headers.get('x-oss-hash-crc64ecma')
                    }
                })
            if not UtilClient.is_unset(ctx.get('md5')) and not UtilClient.is_unset(response.headers.get('content-md5')) and not StringClient.equals(ctx.get('md5'), response.headers.get('content-md5')):
                raise TeaException({
                    'code': 'MD5NotMatched',
                    'data': {
                        'clientMD5': ctx.get('md5'),
                        'serverMD5': response.headers.get('content-md5')
                    }
                })
        if not UtilClient.is_unset(response.body):
            if UtilClient.equal_number(response.status_code, 204):
                UtilClient.read_as_string(response.body)
            elif StringClient.equals(request.body_type, 'xml'):
                body_str = UtilClient.read_as_string(response.body)
                response.deserialized_body = body_str
                if not UtilClient.empty(body_str):
                    result = OSS_UtilClient.parse_xml(body_str, request.action)
                    # for no util language
                    # var result : any = XML.parseXml(bodyStr, null);
                    try:
                        response.deserialized_body = UtilClient.assert_as_map(result)
                    except Exception as error:
                        response.deserialized_body = result
            elif UtilClient.equal_string(request.body_type, 'binary'):
                response.deserialized_body = response.body
            elif UtilClient.equal_string(request.body_type, 'byte'):
                byt = UtilClient.read_as_bytes(response.body)
                response.deserialized_body = byt
            elif UtilClient.equal_string(request.body_type, 'string'):
                response.deserialized_body = UtilClient.read_as_string(response.body)
            elif UtilClient.equal_string(request.body_type, 'json'):
                obj = UtilClient.read_as_json(response.body)
                res = UtilClient.assert_as_map(obj)
                response.deserialized_body = res
            elif UtilClient.equal_string(request.body_type, 'array'):
                response.deserialized_body = UtilClient.read_as_json(response.body)
            else:
                response.deserialized_body = UtilClient.read_as_string(response.body)

    async def modify_response_async(
        self,
        context: spi_models.InterceptorContext,
        attribute_map: spi_models.AttributeMap,
    ) -> None:
        request = context.request
        response = context.response
        body_str = None
        if UtilClient.is_4xx(response.status_code) or UtilClient.is_5xx(response.status_code):
            body_str = await UtilClient.read_as_string_async(response.body)
            if not UtilClient.empty(body_str):
                resp_map = XMLClient.parse_xml(body_str, None)
                err = UtilClient.assert_as_map(resp_map.get('Error'))
                raise TeaException({
                    'code': err.get('Code'),
                    'message': err.get('Message'),
                    'data': {
                        'statusCode': response.status_code,
                        'requestId': err.get('RequestId'),
                        'ecCode': err.get('EC'),
                        'Recommend': err.get('RecommendDoc'),
                        'hostId': err.get('HostId'),
                        'AccessDeniedDetail': err.get('AccessDeniedDetail')
                    }
                })
            else:
                headers = response.headers
                request_id = headers.get('x-oss-request-id')
                ec_code = headers.get('x-oss-ec-code')
                raise TeaException({
                    'code': response.status_code,
                    'message': None,
                    'data': {
                        'statusCode': response.status_code,
                        'requestId': f'{request_id}',
                        'ecCode': ec_code
                    }
                })
        ctx = attribute_map.key
        if not UtilClient.is_unset(ctx):
            if not UtilClient.is_unset(ctx.get('crc')) and not UtilClient.is_unset(response.headers.get('x-oss-hash-crc64ecma')) and not StringClient.equals(ctx.get('crc'), response.headers.get('x-oss-hash-crc64ecma')):
                raise TeaException({
                    'code': 'CrcNotMatched',
                    'data': {
                        'clientCrc': ctx.get('crc'),
                        'serverCrc': response.headers.get('x-oss-hash-crc64ecma')
                    }
                })
            if not UtilClient.is_unset(ctx.get('md5')) and not UtilClient.is_unset(response.headers.get('content-md5')) and not StringClient.equals(ctx.get('md5'), response.headers.get('content-md5')):
                raise TeaException({
                    'code': 'MD5NotMatched',
                    'data': {
                        'clientMD5': ctx.get('md5'),
                        'serverMD5': response.headers.get('content-md5')
                    }
                })
        if not UtilClient.is_unset(response.body):
            if UtilClient.equal_number(response.status_code, 204):
                await UtilClient.read_as_string_async(response.body)
            elif StringClient.equals(request.body_type, 'xml'):
                body_str = await UtilClient.read_as_string_async(response.body)
                response.deserialized_body = body_str
                if not UtilClient.empty(body_str):
                    result = await OSS_UtilClient.parse_xml_async(body_str, request.action)
                    # for no util language
                    # var result : any = XML.parseXml(bodyStr, null);
                    try:
                        response.deserialized_body = UtilClient.assert_as_map(result)
                    except Exception as error:
                        response.deserialized_body = result
            elif UtilClient.equal_string(request.body_type, 'binary'):
                response.deserialized_body = response.body
            elif UtilClient.equal_string(request.body_type, 'byte'):
                byt = await UtilClient.read_as_bytes_async(response.body)
                response.deserialized_body = byt
            elif UtilClient.equal_string(request.body_type, 'string'):
                response.deserialized_body = await UtilClient.read_as_string_async(response.body)
            elif UtilClient.equal_string(request.body_type, 'json'):
                obj = await UtilClient.read_as_json_async(response.body)
                res = UtilClient.assert_as_map(obj)
                response.deserialized_body = res
            elif UtilClient.equal_string(request.body_type, 'array'):
                response.deserialized_body = await UtilClient.read_as_json_async(response.body)
            else:
                response.deserialized_body = await UtilClient.read_as_string_async(response.body)

    def get_region_id_from_endpoint(
        self,
        endpoint: str,
    ) -> str:
        if not UtilClient.empty(endpoint):
            idx = -1
            if StringClient.has_prefix(endpoint, 'oss-') and StringClient.has_suffix(endpoint, '.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.aliyuncs.com')
                return StringClient.sub_string(endpoint, 4, idx)
            if StringClient.has_suffix(endpoint, '.mgw.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.mgw.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '.mgw-internal.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.mgw-internal.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '-internal.oss-data-acc.aliyuncs.com'):
                idx = StringClient.index(endpoint, '-internal.oss-data-acc.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '.oss-dls.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.oss-dls.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
        return ''

    async def get_region_id_from_endpoint_async(
        self,
        endpoint: str,
    ) -> str:
        if not UtilClient.empty(endpoint):
            idx = -1
            if StringClient.has_prefix(endpoint, 'oss-') and StringClient.has_suffix(endpoint, '.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.aliyuncs.com')
                return StringClient.sub_string(endpoint, 4, idx)
            if StringClient.has_suffix(endpoint, '.mgw.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.mgw.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '.mgw-internal.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.mgw-internal.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '-internal.oss-data-acc.aliyuncs.com'):
                idx = StringClient.index(endpoint, '-internal.oss-data-acc.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
            if StringClient.has_suffix(endpoint, '.oss-dls.aliyuncs.com'):
                idx = StringClient.index(endpoint, '.oss-dls.aliyuncs.com')
                return StringClient.sub_string(endpoint, 0, idx)
        return ''

    def get_endpoint(
        self,
        region_id: str,
        network: str,
        endpoint: str,
    ) -> str:
        if not UtilClient.empty(endpoint):
            return endpoint
        if UtilClient.empty(region_id):
            region_id = 'cn-hangzhou'
        if not UtilClient.empty(network):
            if StringClient.contains(network, 'internal'):
                return f'oss-{region_id}-internal.aliyuncs.com'
            elif StringClient.contains(network, 'ipv6'):
                return f'{region_id}oss.aliyuncs.com'
            elif StringClient.contains(network, 'accelerate'):
                return f'oss-{network}.aliyuncs.com'
        return f'oss-{region_id}.aliyuncs.com'

    async def get_endpoint_async(
        self,
        region_id: str,
        network: str,
        endpoint: str,
    ) -> str:
        if not UtilClient.empty(endpoint):
            return endpoint
        if UtilClient.empty(region_id):
            region_id = 'cn-hangzhou'
        if not UtilClient.empty(network):
            if StringClient.contains(network, 'internal'):
                return f'oss-{region_id}-internal.aliyuncs.com'
            elif StringClient.contains(network, 'ipv6'):
                return f'{region_id}oss.aliyuncs.com'
            elif StringClient.contains(network, 'accelerate'):
                return f'oss-{network}.aliyuncs.com'
        return f'oss-{region_id}.aliyuncs.com'

    def get_host(
        self,
        endpoint_type: str,
        bucket_name: str,
        endpoint: str,
        context: spi_models.InterceptorContext,
    ) -> str:
        if StringClient.contains(endpoint, '.mgw.aliyuncs.com') and not UtilClient.is_unset(context.request.host_map.get('userid')):
            return f"{context.request.host_map.get('userid')}.{endpoint}"
        if StringClient.contains(endpoint, '.mgw-internal.aliyuncs.com') and not UtilClient.is_unset(context.request.host_map.get('userid')):
            return f"{context.request.host_map.get('userid')}.{endpoint}"
        if UtilClient.empty(bucket_name):
            return endpoint
        host = f'{bucket_name}.{endpoint}'
        if not UtilClient.empty(endpoint_type):
            if StringClient.equals(endpoint_type, 'ip'):
                host = f'{endpoint}/{bucket_name}'
            elif StringClient.equals(endpoint_type, 'cname'):
                host = endpoint
        return host

    async def get_host_async(
        self,
        endpoint_type: str,
        bucket_name: str,
        endpoint: str,
        context: spi_models.InterceptorContext,
    ) -> str:
        if StringClient.contains(endpoint, '.mgw.aliyuncs.com') and not UtilClient.is_unset(context.request.host_map.get('userid')):
            return f"{context.request.host_map.get('userid')}.{endpoint}"
        if StringClient.contains(endpoint, '.mgw-internal.aliyuncs.com') and not UtilClient.is_unset(context.request.host_map.get('userid')):
            return f"{context.request.host_map.get('userid')}.{endpoint}"
        if UtilClient.empty(bucket_name):
            return endpoint
        host = f'{bucket_name}.{endpoint}'
        if not UtilClient.empty(endpoint_type):
            if StringClient.equals(endpoint_type, 'ip'):
                host = f'{endpoint}/{bucket_name}'
            elif StringClient.equals(endpoint_type, 'cname'):
                host = endpoint
        return host

    def get_authorization(
        self,
        signature_version: str,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        ak: str,
        secret: str,
        region_id: str,
    ) -> str:
        sign = ''
        if not UtilClient.is_unset(signature_version):
            if StringClient.equals(signature_version, 'v1'):
                sign = self.get_signature_v1(bucket_name, pathname, method, query, headers, secret)
                return f'OSS {ak}:{sign}'
            if StringClient.equals(signature_version, 'v2'):
                sign = self.get_signature_v2(bucket_name, pathname, method, query, headers, secret)
                return f'OSS2 AccessKeyId:{ak},Signature:{sign}'
        date_time = OpenApiUtilClient.get_timestamp()
        date_time = StringClient.replace(date_time, '-', '', None)
        date_time = StringClient.replace(date_time, ':', '', None)
        headers['x-oss-date'] = date_time
        headers['x-oss-content-sha256'] = 'UNSIGNED-PAYLOAD'
        only_date = StringClient.sub_string(date_time, 0, 8)
        cred = f'{ak}/{only_date}/{region_id}/oss/aliyun_v4_request'
        sign = self.get_signature_v4(bucket_name, pathname, method, query, headers, only_date, region_id, secret)
        return f'OSS4-HMAC-SHA256 Credential={cred}, Signature={sign}'

    async def get_authorization_async(
        self,
        signature_version: str,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        ak: str,
        secret: str,
        region_id: str,
    ) -> str:
        sign = ''
        if not UtilClient.is_unset(signature_version):
            if StringClient.equals(signature_version, 'v1'):
                sign = await self.get_signature_v1_async(bucket_name, pathname, method, query, headers, secret)
                return f'OSS {ak}:{sign}'
            if StringClient.equals(signature_version, 'v2'):
                sign = await self.get_signature_v2_async(bucket_name, pathname, method, query, headers, secret)
                return f'OSS2 AccessKeyId:{ak},Signature:{sign}'
        date_time = OpenApiUtilClient.get_timestamp()
        date_time = StringClient.replace(date_time, '-', '', None)
        date_time = StringClient.replace(date_time, ':', '', None)
        headers['x-oss-date'] = date_time
        headers['x-oss-content-sha256'] = 'UNSIGNED-PAYLOAD'
        only_date = StringClient.sub_string(date_time, 0, 8)
        cred = f'{ak}/{only_date}/{region_id}/oss/aliyun_v4_request'
        sign = await self.get_signature_v4_async(bucket_name, pathname, method, query, headers, only_date, region_id, secret)
        return f'OSS4-HMAC-SHA256 Credential={cred}, Signature={sign}'

    def get_sign_key(
        self,
        secret: str,
        only_date: str,
        region_id: str,
    ) -> bytes:
        temp = f'aliyun_v4{secret}'
        res = Signer.hmac_sha256sign(only_date, temp)
        res = Signer.hmac_sha256sign_by_bytes(region_id, res)
        res = Signer.hmac_sha256sign_by_bytes('oss', res)
        res = Signer.hmac_sha256sign_by_bytes('aliyun_v4_request', res)
        return res

    async def get_sign_key_async(
        self,
        secret: str,
        only_date: str,
        region_id: str,
    ) -> bytes:
        temp = f'aliyun_v4{secret}'
        res = Signer.hmac_sha256sign(only_date, temp)
        res = Signer.hmac_sha256sign_by_bytes(region_id, res)
        res = Signer.hmac_sha256sign_by_bytes('oss', res)
        res = Signer.hmac_sha256sign_by_bytes('aliyun_v4_request', res)
        return res

    def get_signature_v4(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        only_date: str,
        region_id: str,
        secret: str,
    ) -> str:
        signingkey = self.get_sign_key(secret, only_date, region_id)
        canonicalized_uri = pathname
        if not UtilClient.empty(pathname):
            if not UtilClient.empty(bucket_name):
                canonicalized_uri = f'/{bucket_name}{canonicalized_uri}'
        else:
            if not UtilClient.empty(bucket_name):
                canonicalized_uri = f'/{bucket_name}/'
            else:
                canonicalized_uri = '/'
        # for java:
        # String suffix = (!canonicalizedUri.equals("/") && canonicalizedUri.endsWith("/"))? "/" : "";
        # canonicalizedUri = com.aliyun.openapiutil.Client.getEncodePath(canonicalizedUri) + suffix;
        canonicalized_uri = OpenApiUtilClient.get_encode_path(canonicalized_uri)
        query_map = {}
        for query_key in MapClient.key_set(query):
            query_value = None
            if not UtilClient.empty(query.get(query_key)):
                query_value = Encoder.percent_encode(query.get(query_key))
                query_value = StringClient.replace(query_value, '+', '%20', None)
            query_key = Encoder.percent_encode(query_key)
            query_key = StringClient.replace(query_key, '+', '%20', None)
            # for go : queryMap[tea.StringValue(queryKey)] = queryValue
            query_map[query_key] = query_value
        canonicalized_query_string = self.build_canonicalized_query_string_v4(query_map)
        canonicalized_headers = self.build_canonicalized_headers_v4(headers)
        payload = 'UNSIGNED-PAYLOAD'
        canonical_request = f'{method}\n{canonicalized_uri}\n{canonicalized_query_string}\n{canonicalized_headers}\n\n{payload}'
        hex = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(canonical_request), 'ACS4-HMAC-SHA256'))
        scope = f'{only_date}/{region_id}/oss/aliyun_v4_request'
        string_to_sign = f"OSS4-HMAC-SHA256\n{headers.get('x-oss-date')}\n{scope}\n{hex}"
        signature = Signer.hmac_sha256sign_by_bytes(string_to_sign, signingkey)
        return Encoder.hex_encode(signature)

    async def get_signature_v4_async(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        only_date: str,
        region_id: str,
        secret: str,
    ) -> str:
        signingkey = await self.get_sign_key_async(secret, only_date, region_id)
        canonicalized_uri = pathname
        if not UtilClient.empty(pathname):
            if not UtilClient.empty(bucket_name):
                canonicalized_uri = f'/{bucket_name}{canonicalized_uri}'
        else:
            if not UtilClient.empty(bucket_name):
                canonicalized_uri = f'/{bucket_name}/'
            else:
                canonicalized_uri = '/'
        # for java:
        # String suffix = (!canonicalizedUri.equals("/") && canonicalizedUri.endsWith("/"))? "/" : "";
        # canonicalizedUri = com.aliyun.openapiutil.Client.getEncodePath(canonicalizedUri) + suffix;
        canonicalized_uri = OpenApiUtilClient.get_encode_path(canonicalized_uri)
        query_map = {}
        for query_key in MapClient.key_set(query):
            query_value = None
            if not UtilClient.empty(query.get(query_key)):
                query_value = Encoder.percent_encode(query.get(query_key))
                query_value = StringClient.replace(query_value, '+', '%20', None)
            query_key = Encoder.percent_encode(query_key)
            query_key = StringClient.replace(query_key, '+', '%20', None)
            # for go : queryMap[tea.StringValue(queryKey)] = queryValue
            query_map[query_key] = query_value
        canonicalized_query_string = await self.build_canonicalized_query_string_v4_async(query_map)
        canonicalized_headers = await self.build_canonicalized_headers_v4_async(headers)
        payload = 'UNSIGNED-PAYLOAD'
        canonical_request = f'{method}\n{canonicalized_uri}\n{canonicalized_query_string}\n{canonicalized_headers}\n\n{payload}'
        hex = Encoder.hex_encode(Encoder.hash(UtilClient.to_bytes(canonical_request), 'ACS4-HMAC-SHA256'))
        scope = f'{only_date}/{region_id}/oss/aliyun_v4_request'
        string_to_sign = f"OSS4-HMAC-SHA256\n{headers.get('x-oss-date')}\n{scope}\n{hex}"
        signature = Signer.hmac_sha256sign_by_bytes(string_to_sign, signingkey)
        return Encoder.hex_encode(signature)

    def build_canonicalized_query_string_v4(
        self,
        query_map: Dict[str, str],
    ) -> str:
        canonicalized_query_string = ''
        if not UtilClient.is_unset(query_map):
            query_array = MapClient.key_set(query_map)
            sorted_query_array = ArrayClient.asc_sort(query_array)
            separator = ''
            for key in sorted_query_array:
                canonicalized_query_string = f'{canonicalized_query_string}{separator}{key}'
                if not UtilClient.empty(query_map.get(key)):
                    canonicalized_query_string = f'{canonicalized_query_string}={query_map.get(key)}'
                separator = '&'
        return canonicalized_query_string

    async def build_canonicalized_query_string_v4_async(
        self,
        query_map: Dict[str, str],
    ) -> str:
        canonicalized_query_string = ''
        if not UtilClient.is_unset(query_map):
            query_array = MapClient.key_set(query_map)
            sorted_query_array = ArrayClient.asc_sort(query_array)
            separator = ''
            for key in sorted_query_array:
                canonicalized_query_string = f'{canonicalized_query_string}{separator}{key}'
                if not UtilClient.empty(query_map.get(key)):
                    canonicalized_query_string = f'{canonicalized_query_string}={query_map.get(key)}'
                separator = '&'
        return canonicalized_query_string

    def build_canonicalized_headers_v4(
        self,
        headers: Dict[str, str],
    ) -> str:
        canonicalized_headers = ''
        headers_array = MapClient.key_set(headers)
        sorted_headers_array = ArrayClient.asc_sort(headers_array)
        for key in sorted_headers_array:
            lower_key = StringClient.to_lower(key)
            if StringClient.has_prefix(lower_key, 'x-oss-') or StringClient.equals(lower_key, 'content-type') or StringClient.equals(lower_key, 'content-md5'):
                canonicalized_headers = f'{canonicalized_headers}{key}:{StringClient.trim(headers.get(key))}\n'
        return canonicalized_headers

    async def build_canonicalized_headers_v4_async(
        self,
        headers: Dict[str, str],
    ) -> str:
        canonicalized_headers = ''
        headers_array = MapClient.key_set(headers)
        sorted_headers_array = ArrayClient.asc_sort(headers_array)
        for key in sorted_headers_array:
            lower_key = StringClient.to_lower(key)
            if StringClient.has_prefix(lower_key, 'x-oss-') or StringClient.equals(lower_key, 'content-type') or StringClient.equals(lower_key, 'content-md5'):
                canonicalized_headers = f'{canonicalized_headers}{key}:{StringClient.trim(headers.get(key))}\n'
        return canonicalized_headers

    def get_signature_v1(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        secret: str,
    ) -> str:
        resource = ''
        string_to_sign = ''
        if not UtilClient.empty(bucket_name):
            resource = f'/{bucket_name}'
        resource = f'{resource}{pathname}'
        canonicalized_resource = self.build_canonicalized_resource(resource, query)
        canonicalized_headers = self.build_canonicalized_headers(headers)
        string_to_sign = f'{method}\n{canonicalized_headers}{canonicalized_resource}'
        return Encoder.base_64encode_to_string(Signer.hmac_sha1sign(string_to_sign, secret))

    async def get_signature_v1_async(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        secret: str,
    ) -> str:
        resource = ''
        string_to_sign = ''
        if not UtilClient.empty(bucket_name):
            resource = f'/{bucket_name}'
        resource = f'{resource}{pathname}'
        canonicalized_resource = await self.build_canonicalized_resource_async(resource, query)
        canonicalized_headers = await self.build_canonicalized_headers_async(headers)
        string_to_sign = f'{method}\n{canonicalized_headers}{canonicalized_resource}'
        return Encoder.base_64encode_to_string(Signer.hmac_sha1sign(string_to_sign, secret))

    def build_canonicalized_resource(
        self,
        pathname: str,
        query: Dict[str, str],
    ) -> str:
        canonicalized_resource = pathname
        query_keys = MapClient.key_set(query)
        sorted_params = ArrayClient.asc_sort(query_keys)
        separator = '?'
        for param_name in sorted_params:
            if ArrayClient.contains(self._default_signed_params, param_name) or StringClient.has_prefix(param_name, 'x-oss-'):
                canonicalized_resource = f'{canonicalized_resource}{separator}{param_name}'
                if not UtilClient.empty(query.get(param_name)):
                    canonicalized_resource = f'{canonicalized_resource}={query.get(param_name)}'
                separator = '&'
        return canonicalized_resource

    async def build_canonicalized_resource_async(
        self,
        pathname: str,
        query: Dict[str, str],
    ) -> str:
        canonicalized_resource = pathname
        query_keys = MapClient.key_set(query)
        sorted_params = ArrayClient.asc_sort(query_keys)
        separator = '?'
        for param_name in sorted_params:
            if ArrayClient.contains(self._default_signed_params, param_name) or StringClient.has_prefix(param_name, 'x-oss-'):
                canonicalized_resource = f'{canonicalized_resource}{separator}{param_name}'
                if not UtilClient.empty(query.get(param_name)):
                    canonicalized_resource = f'{canonicalized_resource}={query.get(param_name)}'
                separator = '&'
        return canonicalized_resource

    def build_canonicalized_headers(
        self,
        headers: Dict[str, str],
    ) -> str:
        canonicalized_headers = ''
        content_type = headers.get('content-type')
        if UtilClient.is_unset(content_type):
            content_type = ''
        content_md_5 = headers.get('content-md5')
        if UtilClient.is_unset(content_md_5):
            content_md_5 = ''
        canonicalized_headers = f"{canonicalized_headers}{content_md_5}\n{content_type}\n{headers.get('date')}\n"
        keys = MapClient.key_set(headers)
        sorted_headers = ArrayClient.asc_sort(keys)
        for header in sorted_headers:
            if StringClient.contains(StringClient.to_lower(header), 'x-oss-') and not UtilClient.is_unset(headers.get(header)):
                canonicalized_headers = f'{canonicalized_headers}{header}:{headers.get(header)}\n'
        return canonicalized_headers

    async def build_canonicalized_headers_async(
        self,
        headers: Dict[str, str],
    ) -> str:
        canonicalized_headers = ''
        content_type = headers.get('content-type')
        if UtilClient.is_unset(content_type):
            content_type = ''
        content_md_5 = headers.get('content-md5')
        if UtilClient.is_unset(content_md_5):
            content_md_5 = ''
        canonicalized_headers = f"{canonicalized_headers}{content_md_5}\n{content_type}\n{headers.get('date')}\n"
        keys = MapClient.key_set(headers)
        sorted_headers = ArrayClient.asc_sort(keys)
        for header in sorted_headers:
            if StringClient.contains(StringClient.to_lower(header), 'x-oss-') and not UtilClient.is_unset(headers.get(header)):
                canonicalized_headers = f'{canonicalized_headers}{header}:{headers.get(header)}\n'
        return canonicalized_headers

    def get_signature_v2(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        secret: str,
    ) -> str:
        return ''

    async def get_signature_v2_async(
        self,
        bucket_name: str,
        pathname: str,
        method: str,
        query: Dict[str, str],
        headers: Dict[str, str],
        secret: str,
    ) -> str:
        return ''