Cloud SQL Resources

Resources for Google Cloud SQL. Available resources include:

• CloudSQLPostgres: A Postgres cluster running on Google Cloud SQL.
• CloudSQLUser: A user for a Cloud SQL Postgres instance.
• CloudSQLDatabase: A database for a Cloud SQL Postgres instance.

Example Usage

Create a Cloud SQL Postgres instance

1
from sqlalchemy import text
2
import launchflow as lf
3

4
# Automatically creates / connects to a Cloud SQL Postgres cluster in your GCP project
5
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")
6

7
# Quick utilities for connecting to SQLAlchemy, Django, and other ORMs
8
engine = postgres.sqlalchemy_engine()
9

10
with engine.connect() as connection:
11
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)
content_copy

Create a Cloud SQL Postgres instance with a custom database

1
import launchflow as lf
2

3
postgres_instance = lf.gcp.CloudSQLPostgres("my-pg-db", include_default_db=False)
4
postgres_db = lf.gcp.CloudSQLDatabase("my-pg-db", cloud_sql_instance=postgres_instance)
5

6
postgres_db.query("SELECT 1")
content_copy

Create a Cloud SQL Postgres instance with a custom user and database

1
import launchflow as lf
2

3
postgres_instance = lf.gcp.CloudSQLPostgres("my-pg-db", include_default_user=False, include_default_db=False)
4
postgres_user = lf.gcp.CloudSQLUser("my-pg-user", cloud_sql_instance=postgres_instance)
5
postgres_db = lf.gcp.CloudSQLDatabase("my-pg-db", cloud_sql_instance=postgres_instance)
6

7
postgres_db.query("SELECT 1", user=postgres_user)
content_copy

CloudSQLPostgres

A Postgres cluster running on Google Cloud SQL.

Like all Resources, this class configures itself across multiple Environments.

For more information see the official documentation.

Example Usage

1
from sqlalchemy import text
2
import launchflow as lf
3

4
# Automatically creates / connects to a Cloud SQL Postgres cluster in your GCP project
5
postgres = lf.gcp.CloudSQLPostgres("my-pg-db")
6

7
# Quick utilities for connecting to SQLAlchemy, Django, and other ORMs
8
engine = postgres.sqlalchemy_engine()
9

10
with engine.connect() as connection:
11
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)
content_copy

initialization

Create a new Cloud SQL Postgres resource.

Args:

• name (str): The name of the Cloud SQL Postgres instance.
• disk_size_gb (int): The size of the disk in GB. Defaults to 10.
• postgres_version (PostgresVersion): The version of Postgres to use. Defaults to PostgresVersion.POSTGRES_15.
• include_default_db (bool): Whether to include a default database. Defaults to True.
• include_default_user (bool): Whether to include a default user. Defaults to True.
• delete_protection (bool): Whether to enable deletion protection. Defaults to False.
• allow_public_access (bool): Whether to allow public access. Default to True for development environments and False for production environments.
• edition (Literal["ENTERPRISE_PLUS", "ENTERPRISE"]): The edition of the Cloud SQL Postgres instance. Defaults to "ENTERPRISE".
• availability_type (Literal["REGIONAL", "ZONAL"]): The availability type of the Cloud SQL Postgres instance. Defaults to "ZONAL" for developments environments and "REGIONAL" for production environments.
• database_tier (Literal["BASIC", "STANDARD_HA", "ENTERPRISE"]): The tier of the Cloud SQL Postgres instance. Defaults to "db-f1-micro" for development environments and "db-custom-1-3840" for production environments.
• database_flags (Dict[Any, Any]): Additional database flags to pass to your database instance. See: https://cloud.google.com/sql/docs/postgres/flags

query

1
CloudSQLPostgres.query(query: str, user: Optional["CloudSQLUser"] = None)
content_copy

Executes a query on the Cloud SQL Postgres instance.

Args:

• query (str): The SQL query to execute.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• The results of the query.

Example usage:

1
import launchflow as lf
2

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

5
# Executes a query on the Cloud SQL Postgres instance
6
postgres.query("SELECT 1")
content_copy

NOTE: This method is not recommended for production use. Use sqlalchemy_engine instead.

django_settings

1
CloudSQLPostgres.django_settings(user: Optional["CloudSQLUser"] = None)
content_copy

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

Args:

• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• A dictionary of Django settings 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
}
content_copy

sqlalchemy_engine_options

1
CloudSQLPostgres.sqlalchemy_engine_options(*, ip_type=None, user: Optional["CloudSQLUser"] = None)
content_copy

Get the SQLAlchemy engine options 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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• The SQLAlchemy engine options.

sqlalchemy_async_engine_options

1
async CloudSQLPostgres.sqlalchemy_async_engine_options(ip_type=None, user: Optional["CloudSQLUser"] = None)
content_copy

Get the async SQLAlchemy engine options 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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• The async SQLAlchemy engine options.

sqlalchemy_engine

1
CloudSQLPostgres.sqlalchemy_engine(*, ip_type=None, user: Optional["CloudSQLUser"] = None, **engine_kwargs)
content_copy

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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.
• **engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Returns:

• The SQLAlchemy 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,)
content_copy

sqlalchemy_async_engine

1
async CloudSQLPostgres.sqlalchemy_async_engine(*, ip_type=None, user: Optional["CloudSQLUser"] = None, **engine_kwargs)
content_copy

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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.
• **engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Returns:

• The async SQLAlchemy 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())
content_copy

CloudSQLUser

initialization

Create a new Cloud SQL User resource.

Args:

• name (str): The name of the Cloud SQL User.
• cloud_sql_instance (CloudSQLPostgres): The Cloud SQL Postgres instance.
• password (Optional[str]): The password for the Cloud SQL User. Defaults to None.

CloudSQLDatabase

initialization

Create a new Cloud SQL Database resource.

Args:

• name (str): The name of the Cloud SQL Database.
• cloud_sql_instance (CloudSQLPostgres): The Cloud SQL Postgres instance.

query

1
CloudSQLDatabase.query(query: str, user: Optional["CloudSQLUser"] = None)
content_copy

Executes a query on the Cloud SQL Database instance.

Args:

• query (str): The SQL query to execute.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Example usage:

1
import launchflow as lf
2

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

5
# Executes a query on the Cloud SQL Database instance
6
postgres.query("SELECT 1")
content_copy

NOTE: This method is not recommended for production use. Use sqlalchemy_engine instead.

django_settings

1
CloudSQLDatabase.django_settings(user: Optional[CloudSQLUser] = None)
content_copy

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

Args:

• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• A dictionary of Django settings 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
}
content_copy

sqlalchemy_engine_options

1
CloudSQLDatabase.sqlalchemy_engine_options(*, ip_type=None, user: Optional[CloudSQLUser] = None)
content_copy

Get the SQLAlchemy engine options 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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• The SQLAlchemy engine options.

sqlalchemy_async_engine_options

1
async CloudSQLDatabase.sqlalchemy_async_engine_options(ip_type=None, user: Optional[CloudSQLUser] = None)
content_copy

Get the async SQLAlchemy engine options 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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.

Returns:

• The async SQLAlchemy engine options.

sqlalchemy_engine

1
CloudSQLDatabase.sqlalchemy_engine(*, ip_type=None, user: Optional[CloudSQLUser] = None, **engine_kwargs)
content_copy

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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.
• **engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Returns:

• The SQLAlchemy 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,)
content_copy

sqlalchemy_async_engine

1
async CloudSQLDatabase.sqlalchemy_async_engine(*, ip_type=None, user: Optional["CloudSQLUser"] = None, **engine_kwargs)
content_copy

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.
• user (CloudSQLUser): The CloudSQLUser to authenticate as. If not provided the default user for the instance will be used.
• **engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Returns:

• The async SQLAlchemy 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())
content_copy