CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-identity

Microsoft Azure Identity Library providing authentication credentials for Azure SDK clients.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

async.mddocs/

Async Support

Asynchronous versions of all azure-identity credentials for use with asyncio-based applications and Azure SDK async clients. All async credentials implement the AsyncTokenCredential protocol and provide the same authentication capabilities as their synchronous counterparts.

Core Async Module

# Import async credentials from azure.identity.aio
from azure.identity.aio import (
    DefaultAzureCredential,
    ClientSecretCredential,
    CertificateCredential,
    ManagedIdentityCredential,
    InteractiveBrowserCredential,
    DeviceCodeCredential,
    AzureCliCredential,
    ChainedTokenCredential
)

Capabilities

AsyncTokenCredential Protocol

All async credentials implement the AsyncTokenCredential protocol, providing asynchronous token acquisition methods.

from abc import ABC, abstractmethod
from typing import Optional, Any
from azure.core.credentials import AccessToken

class AsyncTokenCredential(ABC):
    @abstractmethod
    async def get_token(self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None, **kwargs: Any) -> AccessToken:
        """
        Asynchronously request an access token for the specified scopes.
        
        Args:
            *scopes: Desired scopes for the access token
            claims: Additional claims required in the token
            tenant_id: Optional tenant ID override
            **kwargs: Additional keyword arguments
            
        Returns:
            AccessToken: The access token with expiration information
        """

    async def get_token_info(self, *scopes: str, options: Optional[dict] = None) -> dict:
        """
        Asynchronously request an access token with additional information.
        
        Args:
            *scopes: Desired scopes for the access token
            options: Additional options for token acquisition
            
        Returns:
            dict: Token information including access token and metadata
        """

    async def close(self) -> None:
        """Close the credential's transport session."""

    async def __aenter__(self):
        """Async context manager entry."""
        return self

    async def __aexit__(self, *args):
        """Async context manager exit."""
        await self.close()

DefaultAzureCredential (Async)

Asynchronous version of the intelligent credential chain that automatically detects available authentication methods.

from azure.identity.aio import DefaultAzureCredential

class DefaultAzureCredential:
    def __init__(self, **kwargs):
        """
        Create an async DefaultAzureCredential with the same parameters as the sync version.
        
        Attempts credentials in the same order as the synchronous version:
        1. EnvironmentCredential
        2. WorkloadIdentityCredential  
        3. ManagedIdentityCredential
        4. SharedTokenCacheCredential
        5. VisualStudioCodeCredential (excluded by default)
        6. AzureCliCredential
        7. AzurePowerShellCredential
        8. AzureDeveloperCliCredential
        9. InteractiveBrowserCredential (excluded by default)
        """

    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
        """Asynchronously request an access token using the credential chain."""

    async def get_token_info(self, *scopes: str, options: Optional[dict] = None) -> dict:
        """Asynchronously request access token with additional information."""

    async def close(self) -> None:
        """Close the credential and its underlying transport."""

Usage Example:

import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient

async def main():
    # Create async credential
    credential = DefaultAzureCredential()
    
    try:
        # Use with async Azure SDK client
        async with BlobServiceClient(
            account_url="https://account.blob.core.windows.net",
            credential=credential
        ) as blob_client:
            
            # Asynchronously list containers
            async for container in blob_client.list_containers():
                print(f"Container: {container.name}")
                
    finally:
        # Always close the credential
        await credential.close()

# Run async application
asyncio.run(main())

Service Principal Credentials (Async)

Asynchronous versions of service principal authentication methods.

from azure.identity.aio import ClientSecretCredential, CertificateCredential, ClientAssertionCredential

# All service principal credentials support the same constructor parameters
# as their synchronous counterparts

class ClientSecretCredential:
    def __init__(self, tenant_id: str, client_id: str, client_secret: str, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class CertificateCredential:
    def __init__(self, tenant_id: str, client_id: str, certificate_path: Optional[str] = None, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class ClientAssertionCredential:
    def __init__(self, tenant_id: str, client_id: str, func: Callable[[], str], **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

Usage Example:

import asyncio
from azure.identity.aio import ClientSecretCredential
from azure.keyvault.secrets.aio import SecretClient

async def access_secrets():
    credential = ClientSecretCredential(
        tenant_id="your-tenant-id",
        client_id="your-client-id",
        client_secret="your-client-secret"
    )
    
    async with credential:  # Use context manager for automatic cleanup
        async with SecretClient(
            vault_url="https://vault.vault.azure.net/",
            credential=credential
        ) as secret_client:
            
            secret = await secret_client.get_secret("database-password")
            print(f"Retrieved secret: {secret.name}")

asyncio.run(access_secrets())

Interactive Credentials (Async)

Asynchronous versions of interactive user authentication methods.

from azure.identity.aio import (
    InteractiveBrowserCredential,
    DeviceCodeCredential,
    UsernamePasswordCredential
)

class InteractiveBrowserCredential:
    def __init__(self, **kwargs): ...
    async def authenticate(self, *scopes: str, **kwargs) -> AuthenticationRecord: ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class DeviceCodeCredential:
    def __init__(self, **kwargs): ...
    async def authenticate(self, *scopes: str, **kwargs) -> AuthenticationRecord: ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class UsernamePasswordCredential:
    def __init__(self, client_id: str, username: str, password: str, **kwargs): ...
    async def authenticate(self, *scopes: str, **kwargs) -> AuthenticationRecord: ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

Usage Example:

import asyncio
from azure.identity.aio import InteractiveBrowserCredential

async def interactive_auth():
    credential = InteractiveBrowserCredential()
    
    try:
        # Perform async authentication
        record = await credential.authenticate(
            scopes=["https://graph.microsoft.com/.default"]
        )
        print(f"Authenticated user: {record.username}")
        
        # Get token asynchronously
        token = await credential.get_token("https://graph.microsoft.com/.default")
        print("Token acquired successfully")
        
    finally:
        await credential.close()

asyncio.run(interactive_auth())

Azure Platform Credentials (Async)

Asynchronous versions of Azure-native authentication methods.

from azure.identity.aio import (
    ManagedIdentityCredential,
    WorkloadIdentityCredential,
    AzurePipelinesCredential
)

class ManagedIdentityCredential:
    def __init__(self, *, client_id: Optional[str] = None, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class WorkloadIdentityCredential:
    def __init__(self, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class AzurePipelinesCredential:
    def __init__(self, *, system_access_token: str, service_connection_id: str, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

Usage Example:

import asyncio
from azure.identity.aio import ManagedIdentityCredential
from azure.storage.blob.aio import BlobServiceClient

async def managed_identity_example():
    # Use managed identity in Azure-hosted application
    credential = ManagedIdentityCredential()
    
    async with credential:
        async with BlobServiceClient(
            account_url="https://account.blob.core.windows.net",
            credential=credential
        ) as blob_client:
            
            containers = []
            async for container in blob_client.list_containers():
                containers.append(container.name)
            
            print(f"Found {len(containers)} containers")

asyncio.run(managed_identity_example())

Developer Tool Credentials (Async)

Asynchronous versions of developer tool authentication methods.

from azure.identity.aio import (
    AzureCliCredential,
    AzureDeveloperCliCredential,
    AzurePowerShellCredential,
    VisualStudioCodeCredential,
    SharedTokenCacheCredential
)

class AzureCliCredential:
    def __init__(self, *, process_timeout: int = 10, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class AzureDeveloperCliCredential:
    def __init__(self, *, process_timeout: int = 10, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class AzurePowerShellCredential:
    def __init__(self, *, process_timeout: int = 10, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class VisualStudioCodeCredential:
    def __init__(self, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

class SharedTokenCacheCredential:
    def __init__(self, username: Optional[str] = None, **kwargs): ...
    async def get_token(self, *scopes: str, **kwargs) -> AccessToken: ...
    async def close(self) -> None: ...

ChainedTokenCredential (Async)

Asynchronous version of the configurable credential chain.

from azure.identity.aio import ChainedTokenCredential

class ChainedTokenCredential:
    def __init__(self, *credentials):
        """
        Create an async credential chain from async credential instances.
        
        Args:
            *credentials: AsyncTokenCredential instances to try in order
        """

    async def get_token(self, *scopes: str, **kwargs) -> AccessToken:
        """Asynchronously request token using first available credential."""

    async def close(self) -> None:
        """Close all credentials in the chain."""

Usage Example:

import asyncio
from azure.identity.aio import (
    ChainedTokenCredential,
    ManagedIdentityCredential,
    AzureCliCredential
)

async def custom_chain_example():
    # Create async credential chain
    credential = ChainedTokenCredential(
        ManagedIdentityCredential(),   # Try managed identity first
        AzureCliCredential()           # Fall back to Azure CLI
    )
    
    try:
        token = await credential.get_token("https://management.azure.com/.default")
        print("Authentication successful with credential chain")
    finally:
        await credential.close()

asyncio.run(custom_chain_example())

Async Utility Functions

Asynchronous versions of utility functions for working with credentials.

from azure.identity.aio import get_bearer_token_provider
from typing import Callable, Coroutine, Any

def get_bearer_token_provider(credential: AsyncTokenCredential, *scopes: str) -> Callable[[], Coroutine[Any, Any, str]]:
    """
    Returns a callable that asynchronously provides a bearer token string.
    
    Args:
        credential: The async credential used to authenticate requests
        *scopes: The scopes required for the bearer token
        
    Returns:
        Callable that returns a coroutine yielding a bearer token string
    """

Usage Example:

import asyncio
import aiohttp
from azure.identity.aio import DefaultAzureCredential, get_bearer_token_provider

async def make_graph_request():
    credential = DefaultAzureCredential()
    token_provider = get_bearer_token_provider(
        credential,
        "https://graph.microsoft.com/.default"
    )
    
    try:
        # Get token asynchronously
        token = await token_provider()
        
        # Use with aiohttp
        headers = {"Authorization": f"Bearer {token}"}
        async with aiohttp.ClientSession() as session:
            async with session.get(
                "https://graph.microsoft.com/v1.0/me",
                headers=headers
            ) as response:
                user_data = await response.json()
                print(f"User: {user_data.get('displayName')}")
                
    finally:
        await credential.close()

asyncio.run(make_graph_request())

Async Patterns and Best Practices

Context Manager Pattern

import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient

async def context_manager_pattern():
    # Use async context managers for automatic cleanup
    async with DefaultAzureCredential() as credential:
        async with BlobServiceClient(
            account_url="https://account.blob.core.windows.net",
            credential=credential
        ) as blob_client:
            
            # Perform operations
            async for container in blob_client.list_containers():
                print(f"Container: {container.name}")
    
    # Credentials and clients are automatically closed

asyncio.run(context_manager_pattern())

Resource Management

import asyncio
from azure.identity.aio import ClientSecretCredential

async def proper_resource_management():
    credential = ClientSecretCredential(
        tenant_id="your-tenant-id",
        client_id="your-client-id",
        client_secret="your-client-secret"
    )
    
    try:
        # Use credential
        token = await credential.get_token("https://management.azure.com/.default")
        print("Token acquired")
        
        # Perform other async operations
        await asyncio.sleep(1)  # Simulate work
        
    except Exception as e:
        print(f"Error: {e}")
    finally:
        # Always close credential to free resources
        await credential.close()

asyncio.run(proper_resource_management())

Concurrent Authentication

import asyncio
from azure.identity.aio import DefaultAzureCredential

async def concurrent_token_requests():
    credential = DefaultAzureCredential()
    
    try:
        # Request multiple tokens concurrently
        tasks = [
            credential.get_token("https://graph.microsoft.com/.default"),
            credential.get_token("https://storage.azure.com/.default"),
            credential.get_token("https://management.azure.com/.default")
        ]
        
        tokens = await asyncio.gather(*tasks)
        print(f"Acquired {len(tokens)} tokens concurrently")
        
    finally:
        await credential.close()

asyncio.run(concurrent_token_requests())

Error Handling in Async Context

import asyncio
from azure.identity.aio import (
    DefaultAzureCredential,
    InteractiveBrowserCredential,
    CredentialUnavailableError,
    AuthenticationRequiredError
)

async def robust_async_authentication():
    try:
        # Try DefaultAzureCredential first
        credential = DefaultAzureCredential()
        token = await credential.get_token("https://graph.microsoft.com/.default")
        print("Automatic authentication successful")
        return credential
        
    except CredentialUnavailableError:
        print("Default authentication unavailable, trying interactive")
        
        # Fall back to interactive authentication
        interactive_credential = InteractiveBrowserCredential()
        try:
            token = await interactive_credential.get_token("https://graph.microsoft.com/.default")
            print("Interactive authentication successful")
            return interactive_credential
            
        except AuthenticationRequiredError as e:
            print(f"Authentication required for scopes: {list(e.scopes)}")
            record = await interactive_credential.authenticate(*e.scopes)
            print(f"Authentication completed for: {record.username}")
            return interactive_credential

async def main():
    credential = None
    try:
        credential = await robust_async_authentication()
        # Use credential for subsequent operations
        
    except Exception as e:
        print(f"Authentication failed: {e}")
    finally:
        if credential:
            await credential.close()

asyncio.run(main())

Integration with FastAPI

from fastapi import FastAPI, Depends, HTTPException
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient
import asyncio

app = FastAPI()

# Shared credential instance
credential = DefaultAzureCredential()

@app.on_event("startup")
async def startup_event():
    """Initialize credential on startup."""
    pass

@app.on_event("shutdown") 
async def shutdown_event():
    """Clean up credential on shutdown."""
    await credential.close()

@app.get("/containers")
async def list_containers():
    """List blob containers using async credential."""
    try:
        async with BlobServiceClient(
            account_url="https://account.blob.core.windows.net",
            credential=credential
        ) as blob_client:
            
            containers = []
            async for container in blob_client.list_containers():
                containers.append(container.name)
            
            return {"containers": containers}
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Migration from Sync to Async

Key Differences

  1. Import Path: Use azure.identity.aio instead of azure.identity
  2. Async Methods: All token acquisition methods are coroutines
  3. Context Managers: Use async with for automatic resource cleanup
  4. Error Handling: Same exception types, but handle in async context
  5. Resource Management: Always call await credential.close() or use context managers

Migration Example

# Before (sync)
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

credential = DefaultAzureCredential()
blob_client = BlobServiceClient(
    account_url="https://account.blob.core.windows.net",
    credential=credential
)

containers = list(blob_client.list_containers())

# After (async)
import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient

async def main():
    async with DefaultAzureCredential() as credential:
        async with BlobServiceClient(
            account_url="https://account.blob.core.windows.net",
            credential=credential
        ) as blob_client:
            
            containers = []
            async for container in blob_client.list_containers():
                containers.append(container)

asyncio.run(main())

Install with Tessl CLI

npx tessl i tessl/pypi-azure-identity

docs

advanced.md

async.md

azure-platform.md

core-credentials.md

developer.md

index.md

interactive.md

service-principal.md

tile.json