使用 SQLAlchemy 进行异步连接

此代码段展示了如何创建 SQLAlchemy `AsyncEngine` 以安全地连接到数据库。它使用 AlloyDB Python 连接器的 `AsyncConnector` 自动处理安全且经过授权的数据库连接。

深入探索

如需查看包含此代码示例的详细文档，请参阅以下内容：

• 使用 AlloyDB 语言连接器进行连接

代码示例

Python

如需向 AlloyDB 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅 为本地开发环境设置身份验证。

import asyncpg
import sqlalchemy
import sqlalchemy.ext.asyncio

from google.cloud.alloydbconnector import AsyncConnector


async def create_sqlalchemy_engine(
    inst_uri: str,
    user: str,
    password: str,
    db: str,
    refresh_strategy: str = "background",
    ip_type: str = "PRIVATE",
) -> 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:

        engine, connector = await create_sqlalchemy_engine(
            inst_uri,
            user,
            password,
            db,
        )
        async with engine.connect() as conn:
            time = await conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
            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 database user name, e.g., postgres
        password (str):
            The database user's password, e.g., secret-password
        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.
        ip_type (str):
            IP type used to connect. One of "PUBLIC", "PRIVATE" or "PSC".
    """
    connector = AsyncConnector(refresh_strategy=refresh_strategy)

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

后续步骤

如需搜索和过滤其他 Google Cloud 产品的代码示例，请参阅Google Cloud 示例浏览器。