def _get_credentials_provider()

in alibabacloud_credentials/provider/profile.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

    def _get_credentials_provider(self, section: Dict) -> ICredentialsProvider:

        config_type = section.get(ac.INI_TYPE)
        if 'access_key' == config_type:
            return StaticAKCredentialsProvider(
                access_key_id=section.get('access_key_id'),
                access_key_secret=section.get('access_key_secret')
            )
        elif 'ram_role_arn' == config_type:
            pre_provider = StaticAKCredentialsProvider(
                access_key_id=section.get('access_key_id'),
                access_key_secret=section.get('access_key_secret')
            )
            return RamRoleArnCredentialsProvider(
                credentials_provider=pre_provider,
                role_arn=section.get('role_arn'),
                role_session_name=section.get('role_session_name'),
                policy=section.get('policy')
            )
        elif 'oidc_role_arn' == config_type:
            return OIDCRoleArnCredentialsProvider(
                role_arn=section.get('role_arn'),
                oidc_provider_arn=section.get('oidc_provider_arn'),
                oidc_token_file_path=section.get('oidc_token_file_path'),
                role_session_name=section.get('role_session_name'),
                policy=section.get('policy')
            )
        elif 'ecs_ram_role' == config_type:
            return EcsRamRoleCredentialsProvider(
                role_name=section.get('role_name')
            )
        elif 'rsa_key_pair' == config_type:
            return RsaKeyPairCredentialsProvider(
                public_key_id=section.get('public_key_id'),
                private_key_file=section.get('private_key_file')
            )
        else:
            raise CredentialException(
                f'unsupported credential type {config_type} from credentials file {self._profile_file}')