Conéctate de forma asíncrona con la autenticación de la base de datos de IAM

En este ejemplo, se muestra cómo conectarse a una instancia con un motor asíncrono de SQLAlchemy y el controlador asyncpg. Usa un objeto de escucha de eventos para proporcionar automáticamente un token de acceso de OAuth2 para la autenticación segura y sin contraseña de la base de datos de IAM.

Muestra de código

Python

Para autenticarte en AlloyDB, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

¿Qué sigue?

Si quieres buscar y filtrar muestras de código para otros productos de Google Cloud , consulta el navegador de muestras deGoogle Cloud .