Connect with IAM database authentication and asyncpg

This snippet demonstrates how to create a secure, asynchronous connection pool using the Python asyncpg driver. It connects to an instance using IAM database authentication, which provides an automatically refreshed OAuth2 access token as the password.

Code sample

Python

To authenticate to AlloyDB, set up Application Default Credentials. For more information, see Set up authentication for a local development environment. 

import asyncpg

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

    # initialize Google Auth credentials
    credentials, _ = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )

    def get_authentication_token() -> str:
        """Get OAuth2 access token to be used for IAM database authentication"""
        # refresh credentials if expired
        if not credentials.valid:
            request = Request()
            credentials.refresh(request)
        return credentials.token

    # ... inside of async context (function)
    async with asyncpg.create_pool(
        user=user,  # your IAM db user, e.g. service-account@project-id.iam
        password=get_authentication_token,  # callable to get fresh OAuth2 token
        host=ip_address,  # your AlloyDB instance IP address
        port=5432,
        database=db,  # your database name
        # Because this connection uses an OAuth2 token as a password, you must
        # require SSL, or better, enforce all clients speak SSL on the server
        # side. This ensures the OAuth2 token is not inadvertantly leaked.
        ssl="require",
    ) as pool:
        # acquire connection from native asyncpg connection pool
        async with pool.acquire() as conn:
            time = await conn.fetchrow("SELECT NOW()")
            print("Current time is ", time[0])

What's next

To search and filter code samples for other Google Cloud products, see the Google Cloud sample browser.