SQLAlchemy-Verbindungspool erstellen

In diesem Snippet wird gezeigt, wie Sie einen SQLAlchemy-Verbindungspool erstellen, um Datenbankverbindungen zu verwalten. Mit diesem Ansatz können Sie sicher eine Verbindung zu Ihrer Datenbankinstanz von Python-Anwendungen aus herstellen, die das SQLAlchemy-Toolkit verwenden.

Weitere Informationen

Eine ausführliche Dokumentation, die dieses Codebeispiel enthält, finden Sie hier:

• Verbindung mit den AlloyDB-Sprachconnectors herstellen

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 sqlalchemy

from google.cloud.alloydbconnector import Connector


def create_sqlalchemy_engine(
    inst_uri: str,
    user: str,
    password: str,
    db: str,
    refresh_strategy: str = "background",
) -> tuple[sqlalchemy.engine.Engine, Connector]:
    """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 = create_sqlalchemy_engine(
            inst_uri,
            user,
            password,
            db,
        )
        with engine.connect() as conn:
            time = conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
            conn.commit()
            curr_time = time[0]
            # do something with query result
            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.
    """
    connector = Connector(refresh_strategy=refresh_strategy)

    # create SQLAlchemy connection pool
    engine = sqlalchemy.create_engine(
        "postgresql+pg8000://",
        creator=lambda: connector.connect(
            inst_uri,
            "pg8000",
            user=user,
            password=password,
            db=db,
        ),
    )
    engine.dialect.description_encoding = None
    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.