IAM 데이터베이스 인증을 사용하여 비동기적으로 연결

이 샘플은 SQLAlchemy 비동기 엔진과 asyncpg 드라이버를 사용하여 인스턴스에 연결하는 방법을 보여줍니다. 이벤트 리스너를 사용하여 안전한 비밀번호 없는 IAM 데이터베이스 인증을 위한 OAuth2 액세스 토큰을 자동으로 제공합니다.

코드 샘플

Python

AlloyDB에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 인증 설정을 참조하세요. 

import google.auth
from google.auth.credentials import Credentials
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(credentials: Credentials) -> 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

    engine = create_async_engine(
        # Equivalent URL:
        # postgresql+asyncpg://<user>:empty@<host>:5432/<db_name>
        sqlalchemy.engine.url.URL.create(
            drivername="postgresql+asyncpg",
            username=user,  # your IAM db user, e.g. service-account@project-id.iam
            password="",  # placeholder to be replaced with OAuth2 token
            host=ip_address,  # your AlloyDB instance IP address
            port=5432,
            database=db_name,  # 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.
        connect_args={"ssl": "require"},
    )

    # set 'do_connect' event listener to replace password with OAuth2 token
    # must use engine.sync_engine as async events are not implemented
    @event.listens_for(engine.sync_engine, "do_connect")
    def auto_iam_authentication(dialect, conn_rec, cargs, cparams) -> None:
        cparams["password"] = get_authentication_token(credentials)

    # use connection from connection pool to query AlloyDB database
    async with engine.connect() as conn:
        result = await conn.execute(sqlalchemy.text("SELECT NOW()"))
        time = result.fetchone()
        print("Current time is ", time[0])

다음 단계

다른 Google Cloud 제품의 코드 샘플을 검색하고 필터링하려면 Google Cloud 샘플 브라우저 참조하기