Créer un pool de connexions asynchrone avec l'authentification IAM pour les bases de données

Cet extrait crée un pool de connexions asynchrones SQLAlchemy à l'aide du connecteur AlloyDB. Utilisez cette méthode pour vous connecter de manière sécurisée à votre instance de base de données avec un utilisateur IAM, sans avoir à gérer les mots de passe de la base de données.

Exemple de code

Python

Pour vous authentifier auprès d'AlloyDB, configurez le service Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer l'authentification pour un environnement de développement 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

Étape suivante

Pour rechercher et filtrer des exemples de code pour d'autres produits Google Cloud , consultez l'explorateur d'exemplesGoogle Cloud .