Skip to content

Add support for async credential providers (aiobotocore compatibility) #50

@mvribeiro75

Description

@mvribeiro75

Problem

The current MSKAuthTokenProvider.generate_auth_token_from_credentials_provider() method only supports synchronous botocore credential providers.
When using aiobotocore with an async credential provider, calling:

credential_provider.load()

synchronously returns a coroutine object instead of actual credentials. This causes:

AttributeError: 'coroutine' object has no attribute 'access_key'

because the returned coroutine is never awaited.


How to Reproduce

Below is a minimal example showing the incompatibility.

import asyncio
from datetime import datetime, timedelta
from botocore.credentials import CredentialProvider
from aiobotocore.credentials import AioRefreshableCredentials
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider

class AsyncMyCredentialProvider(CredentialProvider):
    CANONICAL_NAME = "async-app"

    async def load(self):
        await asyncio.sleep(0.1)  # Simulate async API call
        return AioRefreshableCredentials(
            access_key="AKIAEXAMPLE123",
            secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
            token="example_session_token",
            expiry_time=datetime.now() + timedelta(hours=1),
            method="async-app",
            refresh_using=self._async_creds_fetcher,
        )

    async def _async_creds_fetcher(self):
        await asyncio.sleep(0.1)
        return {
            "access_key": "AKIAEXAMPLE123",
            "secret_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
            "token": "example_session_token",
        }

# Attempt to use async provider
async_provider = AsyncMyCredentialProvider()
region = "us-west-2"

token, expiry = MSKAuthTokenProvider.generate_auth_token_from_credentials_provider(
    region, async_provider
)

Result:

AttributeError: 'coroutine' object has no attribute 'access_key'

Why this happens:

  1. generate_auth_token_from_credentials_provider() calls credential_provider.load() synchronously.
  2. For async providers, .load() is a coroutine function, so calling it without await returns a coroutine object.
  3. The MSK signer immediately tries to access .access_key on the coroutine object → boom 💥.

Current Workaround

You can wrap the async provider in a synchronous one:

from botocore.credentials import Credentials

class SyncWrapperProvider(CredentialProvider):
    def __init__(self, async_provider):
        self.async_provider = async_provider

    def load(self):
        creds = asyncio.run(self.async_provider.load())
        return Credentials(
            creds.access_key,
            creds.secret_key,
            creds.token
        )

Then pass SyncWrapperProvider to the MSK signer.


Proposed Solution

Add native support for async credential providers in MSKAuthTokenProvider (or provide an async variant, e.g., generate_auth_token_from_async_credentials_provider) so that applications using aiobotocore do not need custom sync wrappers.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions