RDSPostgres

A Postgres cluster running on AWS's RDS service.

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 RDS Postgres cluster in your AWS account
5
postgres = lf.aws.RDSPostgres("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 RDS Postgres resource.

Args:

• name (str): The name of the RDS Postgres cluster.
• allocated_storage_gb (int): The amount of storage to allocate for the cluster in GB. Defaultus to 20 GB.
• highly_available (Optional[bool]): Whether the database should be made available in multiple availability zones. Defaults to False for development environments and True for production.
• instance_class (Optional[str]): The instance class to use for the RDS Postgres cluster. Defaults to db.t4g.micro for development environments and db.r5.large for production.
• publicly_accessible (Optionally[bool]): Whether the database should be publicly accessible. Defaults to True for development environments and False for production.
• postgres_version (PostgresVersion): The version of Postgres to use. Defaults to PostgresVersion.POSTGRES16.

query

1
RDSPostgres.query(query: str)
content_copy

Executes a SQL query on the Postgres instance running on the RDS cluster.

Args:

• query (str): The SQL query to execute.

Returns:

• The result of the query if it returns rows, otherwise None.

Example usage:

1
import launchflow as lf
2

3
postgres = lf.aws.RDSPostgres("my-pg-db")
4

5
# Executes a query on the Postgres instance running on the RDS cluster
6
postgres.query("SELECT 1")
content_copy

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

django_settings

1
RDSPostgres.django_settings()
content_copy

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

Example usage:

1
import launchflow as lf
2

3
postgres = lf.aws.RDSPostgres("my-pg-db")
4

5
# settings.py
6
DATABASES = {
7
    # Connect Django's ORM to the RDS Postgres instance
8
    "default": postgres.django_settings(),
9
}
content_copy

sqlalchemy_engine

1
RDSPostgres.sqlalchemy_engine(**engine_kwargs)
content_copy

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:

1
import launchflow as lf
2

3
postgres = lf.aws.RDSPostgres("my-pg-db")
4

5
# Creates a SQLAlchemy engine for connecting to the RDS 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 RDSPostgres.sqlalchemy_async_engine(**engine_kwargs)
content_copy

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:

1
import launchflow as lf
2

3
postgres = lf.aws.RDSPostgres("my-pg-db")
4

5
# Creates an async SQLAlchemy engine for connecting to the RDS SQL Postgres instance
6
engine = await postgres.sqlalchemy_async_engine()
7

8
async with engine.connect() as connection:
9
    result = await connection.execute("SELECT 1")
10
    print(await result.fetchone())
content_copy