RDS

A class for creating an RDS instance in AWS.

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

For more information see the official documentation.

Example Usage

1
import launchflow as lf
2

3
# Automatically creates / connects to an RDS cluster in your AWS account
4
rds_instance = lf.aws.RDS("my-db", engine_version=lf.aws.RDSEngineVersion.POSTGRES16)
content_copy

initialization

Create a new RDS resource.

Args:

• name (str): The name of the RDS cluster.
• allocated_storage_gb (int): The amount of storage to allocate for the cluster in GB. Defaults 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 cluster. Defaults to db.t4g.micro for development environments and db.r5.large for production.
• publicly_accessible (Optional[bool]): Whether the database should be publicly accessible. Defaults to True for development environments and False for production.
• engine_version (RDSEngineVersion): The engine version to use. Defaults to RDSEngineVersion.POSTGRES16.

query

1
RDS.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.RDS("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.

sqlalchemy_engine

1
RDS.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.RDS("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