Asynchronen Verbindungspool mit einer öffentlichen IP-Adresse erstellen

Mit diesem Snippet wird ein asynchroner SQLAlchemy-Verbindungspool erstellt, um über die öffentliche IP-Adresse eine Verbindung zu einer Datenbankinstanz herzustellen. Diese Methode bietet eine sichere, verschlüsselte Verbindung, ohne dass Sie SSL/TLS-Zertifikate manuell verwalten müssen.

Codebeispiel

Python

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei AlloyDB zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

import pytest
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,
) -> 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_name (str):
            The name of the database, e.g., mydb
    """
    connector = AsyncConnector()

    # 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="PUBLIC",
        ),
        execution_options={"isolation_level": "AUTOCOMMIT"},
    )
    return engine, connector

Nächste Schritte

Wenn Sie nach Codebeispielen für andere Produkte von Google Cloud suchen und filtern möchten, können Sie den Beispielbrowser fürGoogle Cloud verwenden.