launchflow.aws.rds

RDSPostgres

1class RDSPostgres(AWSResource[RDSPostgresConnectionInfo], PostgresClient)

A Postgres cluster running on AWS's RDS service.

Example usage:

1from sqlalchemy import text
2import launchflow as lf
3
4postgres = lf.aws.RDSPostgres("my-pg-db")
5
6# Quick utilities for connecting to SQLAlchemy, Django, and other ORMs
7engine = postgres.sqlalchemy_engine()
8
9with engine.connect() as connection:
10    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)

init

1def __init__(name: str) -> None

Create a new RDS Postgres resource.

Args:

name: The name of the RDS Postgres cluster.

django_settings

1def django_settings()

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

Example usage:

1import launchflow as lf
2
3postgres = lf.aws.RDSPostgres("my-pg-db")
4
5# settings.py
6DATABASES = {
7    # Connect Django's ORM to the RDS Postgres instance
8    "default": postgres.django_settings(),
9}

sqlalchemy_engine

1def sqlalchemy_engine(**engine_kwargs)

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

Args:

**engine_kwargs: Additional keyword arguments to pass to sqlalchemy.create_engine.

Example usage:

1import launchflow as lf
2
3postgres = lf.aws.RDSPostgres("my-pg-db")
4
5# Creates a SQLAlchemy engine for connecting to the RDS SQL Postgres instance
6engine = postgres.sqlalchemy_engine()
7
8with engine.connect() as connection:
9    print(connection.execute("SELECT 1").fetchone())  # prints (1,)

sqlalchemy_async_engine

1async def sqlalchemy_async_engine(**engine_kwargs)

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

Args:

**engine_kwargs: Additional keyword arguments to pass to create_async_engine.

Example usage:

1import launchflow as lf
2
3postgres = lf.aws.RDSPostgres("my-pg-db")
4
5# Creates an async SQLAlchemy engine for connecting to the RDS SQL Postgres instance
6engine = await postgres.sqlalchemy_async_engine()
7
8async with engine.connect() as connection:
9    result = await connection.execute("SELECT 1")
10    print(await result.fetchone())