launchflow.gcp.cloudsql

CloudSQLPostgres

1
class CloudSQLPostgres(Resource[CloudSQLPostgresConnectionInfo])

A Cloud SQL Postgres resource.

Args:

  • name: The name of the Cloud SQL Postgres instance.

Attributes:

  • connection_name: The connection name of the Cloud SQL Postgres instance. This is available after the resource is created.

Example usage:

1
import launchflow as lf
2
3
db = lf.gcp.CloudSQLPostgres("my-pg-db")

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
db = lf.gcp.CloudSQLPostgres("my-pg-db")
4
engine = db.sqlalchemy_engine()

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
db = lf.gcp.CloudSQLPostgres("my-pg-db")
4
engine = await db.sqlalchemy_async_engine()