tessl/pypi-google-auth

Google Authentication Library providing comprehensive authentication mechanisms for Google APIs and services including OAuth 2.0, JWT, and service account credentials

—

Pending

Overview

Eval results

Files

Async Support

Name: tessl/pypi-google-auth
Author: tessl

Full async implementations using aiohttp for non-blocking authentication flows. Provides async versions of core authentication components enabling high-performance concurrent applications.

Capabilities

Async Credentials

Async versions of core credential classes supporting non-blocking token refresh and authentication operations.

class Credentials(google.auth.credentials.Credentials):
    """Async credentials base class."""
    
    async def refresh(self, request):
        """
        Async refresh of the access token.
        
        Args:
            request (google.auth.aio.transport.Request): Async HTTP transport
            
        Raises:
            google.auth.exceptions.RefreshError: If credentials cannot be refreshed
        """
    
    async def apply(self, headers, token=None):
        """
        Async apply authentication headers to a request.
        
        Args:
            headers (Mapping[str, str]): The HTTP request headers
            token (str): Optional token to use instead of current token
        """

class AnonymousCredentials(google.auth.credentials.AnonymousCredentials):
    """Async anonymous credentials."""
    
    async def refresh(self, request):
        """No-op refresh for anonymous credentials."""

Async Transport

Async HTTP transport implementations using aiohttp for non-blocking requests and credential refresh.

class Request(google.auth.transport.Request):
    """Async HTTP transport using aiohttp."""
    
    def __init__(self, session=None):
        """
        Initialize async transport.
        
        Args:
            session (aiohttp.ClientSession): The aiohttp session.
                If not specified, a new session will be created for each request.
        """
    
    async def __call__(self, method, url, data=None, headers=None, **kwargs):
        """
        Make an async HTTP request.
        
        Args:
            method (str): The HTTP method (GET, POST, etc.)
            url (str): The URL to request
            data (bytes): The request body data
            headers (Mapping[str, str]): Request headers
            **kwargs: Additional arguments to pass to aiohttp
            
        Returns:
            google.auth.transport.Response: The HTTP response
        """

class AuthorizedSession(aiohttp.ClientSession):
    """Async session with automatic credential refresh."""
    
    def __init__(
        self,
        credentials,
        refresh_status_codes=None,
        max_refresh_attempts=None,
        **kwargs
    ):
        """
        Initialize async authorized session.
        
        Args:
            credentials (google.auth.credentials.Credentials): The credentials to use
            refresh_status_codes (Sequence[int]): HTTP status codes that trigger refresh
            max_refresh_attempts (int): Maximum number of refresh attempts
            **kwargs: Additional arguments to pass to aiohttp.ClientSession
        """
    
    async def request(self, method, url, **kwargs):
        """
        Make an async authenticated request.
        
        Args:
            method (str): The HTTP method
            url (str): The URL to request
            **kwargs: Additional arguments to pass to aiohttp
            
        Returns:
            aiohttp.ClientResponse: The HTTP response
        """
    
    async def get(self, url, **kwargs):
        """Make an async GET request."""
    
    async def post(self, url, **kwargs):
        """Make an async POST request."""
    
    async def put(self, url, **kwargs):
        """Make an async PUT request."""
    
    async def delete(self, url, **kwargs):
        """Make an async DELETE request."""

Usage example:

import asyncio
import google.auth
from google.auth.aio.transport import aiohttp

async def main():
    # Get default credentials
    credentials, project = google.auth.default(
        scopes=['https://www.googleapis.com/auth/cloud-platform']
    )
    
    # Create async authorized session
    async with aiohttp.AuthorizedSession(credentials) as session:
        # Make async authenticated requests
        async with session.get('https://www.googleapis.com/compute/v1/projects') as response:
            data = await response.json()
            print(f"Projects: {len(data.get('items', []))}")
        
        # Make multiple concurrent requests
        urls = [
            'https://www.googleapis.com/compute/v1/projects',
            'https://www.googleapis.com/storage/v1/b',
            'https://bigquery.googleapis.com/bigquery/v2/projects'
        ]
        
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        
        for response in responses:
            data = await response.json()
            print(f"Response: {response.status}")

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

Async Service Account Credentials

Async service account authentication supporting non-blocking token refresh and JWT signing.

class Credentials(google.oauth2.service_account.Credentials):
    """Async service account credentials."""
    
    async def refresh(self, request):
        """
        Async refresh of service account token.
        
        Args:
            request (google.auth.aio.transport.Request): Async HTTP transport
        """
    
    @classmethod
    def from_service_account_file(cls, filename, **kwargs):
        """
        Create async service account credentials from file.
        
        Args:
            filename (str): Path to service account JSON file
            **kwargs: Additional arguments
            
        Returns:
            Credentials: Async service account credentials
        """
    
    @classmethod
    def from_service_account_info(cls, info, **kwargs):
        """
        Create async service account credentials from info.
        
        Args:
            info (Mapping[str, str]): Service account info dictionary
            **kwargs: Additional arguments
            
        Returns:
            Credentials: Async service account credentials
        """

Async OAuth2 Credentials

Async OAuth2 user credentials with non-blocking token refresh.

class Credentials(google.oauth2.credentials.Credentials):
    """Async OAuth2 user credentials."""
    
    async def refresh(self, request):
        """
        Async refresh of OAuth2 token using refresh token.
        
        Args:
            request (google.auth.aio.transport.Request): Async HTTP transport
        """

Async JWT Operations

Async JWT token operations for non-blocking JWT creation and verification.

class Credentials(google.auth.jwt.Credentials):
    """Async JWT credentials."""
    
    async def refresh(self, request):
        """
        Async refresh of JWT token.
        
        Args:
            request (google.auth.aio.transport.Request): Async HTTP transport
        """

async def verify_token_async(id_token, request, audience=None, **kwargs):
    """
    Async verify an ID token.
    
    Args:
        id_token (Union[str, bytes]): The encoded token
        request (google.auth.aio.transport.Request): Async HTTP transport
        audience (str): Expected audience
        **kwargs: Additional verification options
        
    Returns:
        Mapping[str, Any]: The decoded token payload
    """

async def fetch_id_token_async(request, audience, service_account_email=None):
    """
    Async fetch an ID token from metadata server.
    
    Args:
        request (google.auth.aio.transport.Request): Async HTTP transport
        audience (str): The audience for the ID token
        service_account_email (str): Service account to impersonate
        
    Returns:
        str: The ID token
    """

Async Reauth

Async reauth flows for multi-factor authentication scenarios.

async def refresh_grant_async(
    request,
    token_uri,
    refresh_token,
    client_id,
    client_secret,
    scopes=None
):
    """
    Async refresh an access token using a refresh token.
    
    Args:
        request (google.auth.aio.transport.Request): Async HTTP transport
        token_uri (str): The OAuth 2.0 token endpoint URI
        refresh_token (str): The refresh token
        client_id (str): The OAuth 2.0 client ID
        client_secret (str): The OAuth 2.0 client secret
        scopes (Sequence[str]): Optional scopes
        
    Returns:
        Tuple[str, Optional[str], Optional[datetime], Mapping[str, str]]:
        Access token, new refresh token, expiry, additional data
    """

Async External Account Credentials

Async external account credentials for workload identity with non-blocking token exchange.

class Credentials(google.auth.external_account.Credentials):
    """Async external account credentials."""
    
    async def refresh(self, request):
        """
        Async refresh using STS token exchange.
        
        Args:
            request (google.auth.aio.transport.Request): Async HTTP transport
        """
    
    async def retrieve_subject_token(self):
        """
        Async retrieve subject token from credential source.
        
        Returns:
            str: The subject token
        """

Async Usage Patterns

Context Manager Pattern

async def api_operations():
    credentials, project = google.auth.default()
    
    # Use async context manager for automatic cleanup
    async with aiohttp.AuthorizedSession(credentials) as session:
        # Session automatically handles credential refresh
        async with session.get('https://api.example.com/data') as response:
            return await response.json()

Concurrent Requests

async def concurrent_api_calls():
    credentials, project = google.auth.default()
    
    async with aiohttp.AuthorizedSession(credentials) as session:
        # Make multiple concurrent authenticated requests
        tasks = [
            session.get('https://api1.googleapis.com/data'),
            session.get('https://api2.googleapis.com/data'),
            session.get('https://api3.googleapis.com/data')
        ]
        
        # Wait for all requests to complete
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        results = []
        for response in responses:
            if isinstance(response, Exception):
                print(f"Request failed: {response}")
            else:
                data = await response.json()
                results.append(data)
        
        return results

Long-Running Services

class AsyncAuthenticatedService:
    def __init__(self):
        self.credentials, self.project = google.auth.default()
        self.session = None
    
    async def start(self):
        """Start the service with persistent session."""
        self.session = aiohttp.AuthorizedSession(self.credentials)
    
    async def stop(self):
        """Clean up resources."""
        if self.session:
            await self.session.close()
    
    async def make_request(self, url):
        """Make authenticated request using persistent session."""
        async with self.session.get(url) as response:
            return await response.json()
    
    async def __aenter__(self):
        await self.start()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.stop()

# Usage
async def main():
    async with AsyncAuthenticatedService() as service:
        result = await service.make_request('https://api.googleapis.com/data')
        print(result)

Error Handling

class RefreshError(google.auth.exceptions.GoogleAuthError):
    """Raised when async credential refresh fails."""
    
class TransportError(google.auth.exceptions.GoogleAuthError):
    """Raised when async transport encounters an error."""

Common async error scenarios: