def __init_

def init()

in alibabacloud_credentials/provider/rsa_key_pair.py [0:0]
45 lines of code
21 McCabe index (conditional complexity)

    def __init__(self, *,
                 public_key_id: str = None,
                 private_key_file: str = None,
                 duration_seconds: int = DEFAULT_DURATION_SECONDS,
                 sts_region_id: str = None,
                 sts_endpoint: str = None,
                 enable_vpc: bool = None,
                 http_options: HttpOptions = None):

        self._public_key_id = public_key_id
        self._private_key_file = private_key_file
        self._duration_seconds = duration_seconds

        if self._duration_seconds is None:
            self._duration_seconds = self.DEFAULT_DURATION_SECONDS
        if self._duration_seconds < 900:
            raise ValueError('session duration should be in the range of 900s - max session duration')
        if self._public_key_id is None or self._public_key_id == '':
            raise ValueError('public_key_id cannot be empty')
        if self._private_key_file is None or self._private_key_file == '':
            raise ValueError('private_key_file cannot be empty')
        self._private_key = _get_content(self._private_key_file)
        if self._private_key is None or self._private_key == '':
            raise ValueError('private_key cannot be empty')

        if sts_endpoint is not None and sts_endpoint != '':
            self._sts_endpoint = sts_endpoint
        else:
            if enable_vpc is not None:
                prefix = 'sts-vpc' if enable_vpc else 'sts'
            else:
                prefix = 'sts-vpc' if au.environment_enable_vpc.lower() == 'true' else 'sts'
            if sts_region_id is not None and sts_region_id != '':
                self._sts_endpoint = f'{prefix}.{sts_region_id}.aliyuncs.com'
            elif au.environment_sts_region is not None and au.environment_sts_region != '':
                self._sts_endpoint = f'{prefix}.{au.environment_sts_region}.aliyuncs.com'
            else:
                self._sts_endpoint = 'sts.ap-northeast-1.aliyuncs.com'

        self._http_options = http_options if http_options is not None else HttpOptions()
        self._runtime_options = {
            'connectTimeout': self._http_options.connect_timeout if self._http_options.connect_timeout is not None else RsaKeyPairCredentialsProvider.DEFAULT_CONNECT_TIMEOUT,
            'readTimeout': self._http_options.read_timeout if self._http_options.read_timeout is not None else RsaKeyPairCredentialsProvider.DEFAULT_READ_TIMEOUT,
            'httpsProxy': self._http_options.proxy
        }
        self._credentials_cache = RefreshCachedSupplier(
            refresh_callable=self._refresh_credentials,
            refresh_callable_async=self._refresh_credentials_async,
        )
def __init__()

def init()
