使用 IAM 数据库身份验证和 asyncpg 进行连接

此代码段演示了如何使用 Python asyncpg 驱动程序创建安全的异步连接池。它使用 IAM 数据库身份验证连接到实例，该身份验证提供自动刷新的 OAuth2 访问令牌作为密码。

代码示例

Python

如需向 AlloyDB 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 

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])

后续步骤

如需搜索和过滤其他 Google Cloud 产品的代码示例，请参阅Google Cloud 示例浏览器。