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

Transport Adapters

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

HTTP transport implementations for making authenticated requests across different HTTP libraries. Transport adapters handle credential application, automatic refresh, and provide consistent interfaces across different HTTP client implementations.

Capabilities

Requests Transport

HTTP transport using the popular requests library, providing session management and automatic credential refresh.

class Request(google.auth.transport.Request):
    """Requests-based HTTP transport for credential refresh."""
    
    def __init__(self, session=None):
        """
        Initialize the transport.
        
        Args:
            session (requests.Session): The requests session to use.
                If not specified, a new session will be created.
        """
    
    def __call__(self, method, url, data=None, headers=None, **kwargs):
        """
        Make an HTTP request.
        
        Args:
            method (str): The HTTP method (GET, POST, etc.)
            url (str): The URL to request
            data (bytes): Request body data
            headers (Mapping[str, str]): Request headers
            **kwargs: Additional arguments to pass to requests
            
        Returns:
            google.auth.transport.Response: The HTTP response
        """

class AuthorizedSession(requests.Session):
    """Requests session with automatic credential refresh."""
    
    def __init__(self, credentials, refresh_status_codes=None, max_refresh_attempts=None):
        """
        Initialize an 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
        """
    
    def request(self, method, url, **kwargs):
        """
        Make an authenticated HTTP request.
        
        Args:
            method (str): HTTP method
            url (str): URL to request
            **kwargs: Additional arguments to pass to requests
            
        Returns:
            requests.Response: The HTTP response
        """

Usage example:

import google.auth
from google.auth.transport import requests

# Get default credentials
credentials, project = google.auth.default()

# Create authenticated session
authed_session = requests.AuthorizedSession(credentials)

# Make authenticated requests
response = authed_session.get('https://www.googleapis.com/compute/v1/projects')
print(response.json())

# Use with different HTTP methods
response = authed_session.post(
    'https://www.googleapis.com/compute/v1/projects/my-project/zones/us-central1-a/instances',
    json={'name': 'test-instance', 'machineType': 'e2-micro'}
)

urllib3 Transport

HTTP transport using urllib3, useful for environments where requests is not available or for lower-level HTTP control.

class Request(google.auth.transport.Request):
    """urllib3-based HTTP transport."""
    
    def __init__(self, http=None):
        """
        Initialize the transport.
        
        Args:
            http (urllib3.PoolManager): The urllib3 HTTP client.
                If not specified, a new PoolManager will be created.
        """

class AuthorizedHttp:
    """urllib3 HTTP client with automatic credential refresh."""
    
    def __init__(self, credentials, http=None, refresh_status_codes=None, max_refresh_attempts=None):
        """
        Initialize an authorized HTTP client.
        
        Args:
            credentials (google.auth.credentials.Credentials): The credentials to use
            http (urllib3.PoolManager): The urllib3 client
            refresh_status_codes (Sequence[int]): Status codes that trigger refresh
            max_refresh_attempts (int): Maximum refresh attempts
        """
    
    def request(self, method, url, **kwargs):
        """
        Make an authenticated HTTP request.
        
        Args:
            method (str): HTTP method
            url (str): URL to request
            **kwargs: Additional arguments to pass to urllib3
            
        Returns:
            urllib3.HTTPResponse: The HTTP response
        """

Usage example:

import google.auth
from google.auth.transport import urllib3

# Get credentials
credentials, project = google.auth.default()

# Create authenticated HTTP client
http = urllib3.AuthorizedHttp(credentials)

# Make requests
response = http.request('GET', 'https://www.googleapis.com/compute/v1/projects')

gRPC Transport

Transport adapter for gRPC-based Google APIs, providing automatic credential refresh and secure channel creation.

def secure_authorized_channel(
    credentials,
    request,
    target,
    ssl_credentials=None,
    client_cert_callback=None,
    **kwargs
):
    """
    Create a secure authorized gRPC channel.
    
    Args:
        credentials (google.auth.credentials.Credentials): The credentials to use
        request (google.auth.transport.Request): HTTP transport for credential refresh
        target (str): The gRPC server address
        ssl_credentials (grpc.ChannelCredentials): SSL credentials for the channel
        client_cert_callback (Callable): Callback for client certificate
        **kwargs: Additional arguments to pass to grpc.secure_channel
        
    Returns:
        grpc.Channel: The secure gRPC channel
    """

class AuthMetadataPlugin(grpc.AuthMetadataPlugin):
    """gRPC authentication metadata plugin."""
    
    def __init__(self, credentials, request):
        """
        Initialize the plugin.
        
        Args:
            credentials (google.auth.credentials.Credentials): The credentials to use
            request (google.auth.transport.Request): HTTP transport for refresh
        """
    
    def __call__(self, context, callback):
        """
        Get authentication metadata for gRPC call.
        
        Args:
            context (grpc.AuthMetadataContext): The gRPC auth context
            callback (grpc.AuthMetadataPluginCallback): Callback for metadata
        """

Usage example:

import google.auth
from google.auth.transport import grpc, requests

# Get credentials
credentials, project = google.auth.default(
    scopes=['https://www.googleapis.com/auth/cloud-platform']
)

# Create request object for credential refresh
request = requests.Request()

# Create secure authorized channel
channel = grpc.secure_authorized_channel(
    credentials,
    request,
    'compute.googleapis.com:443'
)

# Use channel with gRPC stubs
# stub = compute_pb2_grpc.InstancesStub(channel)

Async Transport (aiohttp)

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

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.
        """
    
    async def __call__(self, method, url, data=None, headers=None, **kwargs):
        """
        Make an async HTTP request.
        
        Args:
            method (str): HTTP method
            url (str): URL to request
            data (bytes): Request body
            headers (Mapping[str, str]): Request headers
            **kwargs: Additional arguments
            
        Returns:
            google.auth.transport.Response: The HTTP response
        """

class AuthorizedSession(aiohttp.ClientSession):
    """Async session with automatic credential refresh."""
    
    def __init__(self, credentials, **kwargs):
        """
        Initialize async authorized session.
        
        Args:
            credentials (google.auth.credentials.Credentials): The credentials
            **kwargs: Additional arguments to pass to aiohttp.ClientSession
        """
    
    async def request(self, method, url, **kwargs):
        """
        Make an async authenticated request.
        
        Args:
            method (str): HTTP method
            url (str): URL to request
            **kwargs: Additional arguments
            
        Returns:
            aiohttp.ClientResponse: The HTTP response
        """

Usage example:

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

async def main():
    # Get credentials
    credentials, project = google.auth.default()
    
    # Create async authorized session
    async with aiohttp.AuthorizedSession(credentials) as session:
        # Make async requests
        async with session.get('https://www.googleapis.com/compute/v1/projects') as response:
            data = await response.json()
            print(data)

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

Transport Interface

class Request(abc.ABC):
    """Abstract base class for HTTP transport."""
    
    @abc.abstractmethod
    def __call__(self, method, url, data=None, headers=None, **kwargs):
        """
        Make an HTTP request.
        
        Args:
            method (str): HTTP method
            url (str): URL to request  
            data (bytes): Request body
            headers (Mapping[str, str]): Request headers
            
        Returns:
            Response: HTTP response object
        """

class Response:
    """HTTP response object."""
    
    @property
    def status(self):
        """int: HTTP status code."""
    
    @property
    def headers(self):
        """Mapping[str, str]: Response headers."""
    
    @property
    def data(self):
        """bytes: Response body."""

Error Handling

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

Common transport error scenarios: