in aws_msk_iam_sasl_signer/MSKAuthTokenProvider.py [0:0]
def generate_auth_token_from_credentials_provider(region,
aws_credentials_provider):
"""
Generates an base64-encoded signed url as auth token to authenticate
with an Amazon MSK cluster using IAM Credentials provided by a
credentials provider.
Args: region (str): The AWS region where the cluster is located.
aws_credentials_provider (botocore.credentials.CredentialProvider): The
credentials provider that provides IAM credentials. Returns: str: A
base64-encoded authorization token.
"""
# Check the type of the credentials provider
if not isinstance(aws_credentials_provider, CredentialProvider):
raise TypeError(
"aws_credentials_provider should be of type "
"botocore.credentials.CredentialProvider "
)
# Load credentials
aws_credentials = __load_credentials_from_aws_credentials_provider__(
aws_credentials_provider
)
return __construct_auth_token(region, aws_credentials)