使用 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 範例瀏覽工具。