LaunchFlow Logo

launchflow.gcp.cloudsql

CloudSQLPostgres

1
class CloudSQLPostgres(GCPResource[CloudSQLPostgresConnectionInfo],

2
                       PostgresClient)

A Postgres cluster running on Google Cloud SQL.

Example usage:

1
from sqlalchemy import text

2
import launchflow as lf

3


4
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")

5


6
# Quick utilities for connecting to SQLAlchemy, Django, and other ORMs

7
engine = postgres.sqlalchemy_engine()

8


9
with engine.connect() as connection:

10
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)

__init__

1
def __init__(

2
        name: str,

3
        *,

4
        postgres_version: PostgresVersion = PostgresVersion.POSTGRES_15

5
) -> None

Create a new Cloud SQL Postgres resource.

Args:

  • name: The name of the Cloud SQL Postgres instance.
  • postgres_version: The version of Postgres to use. Defaults to PostgresVersion.POSTGRES_15.

django_settings

1
def django_settings()

Returns a Django settings dictionary for connecting to the Cloud SQL Postgres instance.

Example usage:

1
import launchflow as lf

2


3
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")

4


5
# settings.py

6
DATABASES = {

7
    # Connect Django's ORM to the Cloud SQL Postgres instance

8
    "default": postgres.django_settings(),

9
}

sqlalchemy_engine

1
def sqlalchemy_engine(*, ip_type=None, **engine_kwargs)

Returns a SQLAlchemy engine for connecting to the Cloud SQL Postgres instance.

Args:

  • ip_type: The IP type to use for the connection. If not provided will default to the most permisive IP address. For example if your Cloud SQL instance is provisioned with a public IP address, the default will be IPTypes.PUBLIC. Otherwise it will default to IPTypes.PRIVATE.
  • **engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Example usage:

1
import launchflow as lf

2


3
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")

4


5
# Creates a SQLAlchemy engine for connecting to the Cloud SQL Postgres instance

6
engine = postgres.sqlalchemy_engine()

7


8
with engine.connect() as connection:

9
    print(connection.execute("SELECT 1").fetchone())  # prints (1,)

sqlalchemy_async_engine

1
async def sqlalchemy_async_engine(*, ip_type=None, **engine_kwargs)

Returns an async SQLAlchemy engine for connecting to the Cloud SQL Postgres instance.

Args:

  • ip_type: The IP type to use for the connection. If not provided will default to the most permisive IP address. For example if your Cloud SQL instance is provisioned with a public IP address, the default will be IPTypes.PUBLIC. Otherwise it will default to IPTypes.PRIVATE. - **engine_kwargs: Additional keyword arguments to pass to create_async_engine.

Example usage:

1
import launchflow as lf

2


3
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")

4


5
# Creates an async SQLAlchemy engine for connecting to the Cloud SQL Postgres instance

6
engine = await postgres.sqlalchemy_async_engine()

7


8
async with engine.begin() as connection:

9
    result = await connection.execute("SELECT 1")

10
    print(await result.fetchone())