Google Cloud Async I/O Authentication Client

5.4.4 · active · verified Sun Mar 29

gcloud-aio-auth is an asyncio-compatible Python client library for Google Cloud Authentication. It provides asynchronous primitives for managing access tokens, IAP tokens, and interacting with IAM. Part of the broader `gcloud-aio` monorepo, it offers async interfaces to various Google Cloud services. The current version is 5.4.4, with releases occurring as part of the actively developed monorepo, often tied to dependency updates or new feature rollouts across components.

Warnings

breaking Python 3.9 support was dropped in gcloud-aio-auth version 5.4.4. Users on Python 3.9 must upgrade their Python version to 3.10 or newer. [cite: auth-5.4.4 release notes]
Fix: Upgrade Python environment to version 3.10 or later.
gotcha The `auto_decompress` parameter in `aiohttp.ClientSession` could be inadvertently overwritten by `gcloud-aio-auth` in versions prior to 5.4.4. If you explicitly configure `auto_decompress` on your `ClientSession`, ensure you are on version 5.4.4 or later. [cite: auth-5.4.4 release notes, 7]
Fix: Upgrade to gcloud-aio-auth>=5.4.4. If manually managing `aiohttp.ClientSession`, ensure `auto_decompress=None` is explicitly set on the `Token` (or other gcloud-aio client) constructor if you want to avoid overwriting your session's setting.
gotcha The `gcloud-aio-auth.Token` class manages credentials specifically for the `gcloud-aio` ecosystem. While related to `google.auth` libraries, direct interoperability by passing `google.auth.default()` credentials objects to `gcloud-aio` clients is not the standard pattern and can lead to issues. Rely on `gcloud-aio.auth.Token` for authentication within `gcloud-aio` clients.
Fix: Always initialize `gcloud.aio.auth.Token` (or `IapToken`) as described in `gcloud-aio-auth` documentation, letting it handle credential discovery or providing a `service_file` directly. Do not attempt to pass `google.auth.default()` credential objects to `gcloud-aio` client constructors.
gotcha When creating `Token` instances (or any `gcloud-aio` client that manages an internal `aiohttp.ClientSession`), it is crucial to properly close them to prevent resource leaks. Use them within an `async with` statement or explicitly call `await token.close()` when done.
Fix: Ensure `Token` instances are closed. The recommended pattern is `async with Token(...) as token:` or, if not using a context manager, explicitly call `await token.close()` before your application exits.

Install

pip install gcloud-aio-auth Install stable version

Imports

Token
```
from gcloud.aio.auth import Token
```
Primary class for managing Google Cloud OAuth 2.0 access tokens.
IapToken
```
from gcloud.aio.auth import IapToken
```
Class for handling OpenID Connect ID tokens for IAP-secured services.
IamClient
```
from gcloud.aio.auth import IamClient
```
Client for interacting with Google Cloud IAM public keys and URL signing.

Quickstart

This quickstart demonstrates how to initialize the `Token` class, which handles credential discovery (via `GOOGLE_APPLICATION_CREDENTIALS` or Application Default Credentials) and automatic token refreshing. It then shows how to use the obtained access token to make an authenticated request using `aiohttp.ClientSession` to a Google Cloud API. Remember to run `gcloud auth application-default login` for local development or set the `GOOGLE_APPLICATION_CREDENTIALS` environment variable.

import asyncio
import os
import aiohttp
from gcloud.aio.auth import Token

async def main():
    # Attempt to use GOOGLE_APPLICATION_CREDENTIALS environment variable or ADC
    # For local development, ensure `gcloud auth application-default login` has been run
    # or GOOGLE_APPLICATION_CREDENTIALS points to a service account key file.
    service_account_path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', '')
    
    print("Initializing Token...")
    # Initialize Token; it will try to discover credentials if service_file is None.
    # Specify necessary scopes for your application.
    token = Token(
        service_file=service_account_path if service_account_path else None,
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    
    async with aiohttp.ClientSession() as session:
        try:
            # Get an access token (automatically refreshed by the Token instance)
            access_token = await token.get()
            print(f"Successfully obtained access token (first 10 chars): {access_token[:10]}...")
            
            # Example: Make an authenticated request to a Google Cloud API
            # (e.g., list buckets in Google Cloud Storage)
            headers = {"Authorization": f"Bearer {access_token}"}
            print("Making a dummy authenticated request to Google Cloud Storage API...")
            async with session.get(
                "https://www.googleapis.com/storage/v1/projects/_/buckets",
                headers=headers
            ) as response:
                if response.status == 200:
                    print(f"Request to GCS successful (status 200). Some buckets (if any): {await response.json()}")
                else:
                    print(f"Request failed with status: {response.status}")
                    print(f"Response body: {await response.text()}")
                    
        except Exception as e:
            print(f"An error occurred: {e}")
        finally:
            # Ensure the token's internal session is closed
            await token.close()

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →