Criar um pool de conexões assíncrono com autenticação de banco de dados do IAM

Este snippet cria um pool de conexões assíncronas do SQLAlchemy usando o conector do AlloyDB. Use esse método para se conectar com segurança à instância do banco de dados com um usuário do IAM, evitando a necessidade de gerenciar senhas de banco de dados.

Exemplo de código

Python

Para autenticar no AlloyDB, configure o Application Default Credentials. Se quiser mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local. 

import sqlalchemy
import sqlalchemy.ext.asyncio

from google.cloud.alloydbconnector import AsyncConnector


async def create_sqlalchemy_engine(
    inst_uri: str, user: str, db: str, refresh_strategy: str = "background"
) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, AsyncConnector]:
    """Creates a connection pool for an AlloyDB instance and returns the pool
    and the connector. Callers are responsible for closing the pool and the
    connector.

    A sample invocation looks like:

        pool, connector = await create_sqlalchemy_engine(
            inst_uri,
            user,
            db,
        )
        async with pool.connect() as conn:
            time = (await conn.execute(sqlalchemy.text("SELECT NOW()"))).fetchone()
            conn.commit()
            curr_time = time[0]
            # do something with query result
            await connector.close()

    Args:
        instance_uri (str):
            The instance URI specifies the instance relative to the project,
            region, and cluster. For example:
            "projects/my-project/locations/us-central1/clusters/my-cluster/instances/my-instance"
        user (str):
            The formatted IAM database username.
            e.g., my-email@test.com, service-account@project-id.iam
        db (str):
            The name of the database, e.g., mydb
        refresh_strategy (Optional[str]):
            Refresh strategy for the AlloyDB Connector. Can be one of "lazy"
            or "background". For serverless environments use "lazy" to avoid
            errors resulting from CPU being throttled.
    """
    connector = AsyncConnector(refresh_strategy=refresh_strategy)

    # create async SQLAlchemy connection pool
    engine = sqlalchemy.ext.asyncio.create_async_engine(
        "postgresql+asyncpg://",
        async_creator=lambda: connector.connect(
            inst_uri,
            "asyncpg",
            user=user,
            db=db,
            enable_iam_auth=True,
        ),
        execution_options={"isolation_level": "AUTOCOMMIT"},
    )
    return engine, connector

A seguir

Para pesquisar e filtrar exemplos de código de outros Google Cloud produtos, consulte a Google Cloud pesquisa de exemplos de código.