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
1from sqlalchemy import text
2import launchflow as lf
3
4# Automatically creates / connects to a RDS Postgres cluster in your AWS account
5postgres = lf.aws.RDSPostgres("my-pg-db")
6
7# Quick utilities for connecting to SQLAlchemy, Django, and other ORMs
8engine = postgres.sqlalchemy_engine()
9
10with engine.connect() as connection:
11 print(connection.execute(text("SELECT 1")).fetchone()) # prints (1,)
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 toFalse
for development environments andTrue
for production.instance_class (Optional[str])
: The instance class to use for the RDS Postgres cluster. Defaults todb.t4g.micro
for development environments anddb.r5.large
for production.publicly_accessible (Optionally[bool])
: Whether the database should be publicly accessible. Defaults toTrue
for development environments andFalse
for production.postgres_version (PostgresVersion)
: The version of Postgres to use. Defaults toPostgresVersion.POSTGRES16
.
inputs
1RDSPostgres.inputs(environment_state: EnvironmentState) -> RDSPostgresInputs
Get the inputs for the RDS Postgres resource.
Args:
environment_state (EnvironmentState)
: The environment to get state for.
Returns:
RDSPostgresInputs
: The inputs for the RDS Postgres resource.
query
1RDSPostgres.query(query: str)
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:
1import launchflow as lf
2
3postgres = lf.aws.RDSPostgres("my-pg-db")
4
5# Executes a query on the Postgres instance running on the RDS cluster
6postgres.query("SELECT 1")
NOTE: This method is not recommended for production use. Use sqlalchemy_engine
instead.
django_settings
1RDSPostgres.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
1RDSPostgres.sqlalchemy_engine(**engine_kwargs)
Returns a SQLAlchemy engine for connecting to the RDS SQL Postgres instance.
Args:
**engine_kwargs
: Additional keyword arguments to pass tosqlalchemy.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 RDSPostgres.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 tocreate_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())