EC2 Resources

Resources for AWS Elastic Container Service. Available resources include:

• EC2: A Compute Engine VM running a Docker Container.
• EC2Postgres: A Postgres instance running on a VM in AWS EC2.
• EC2Redis: A Redis instance running on a VM on a VM in AWS EC2.

Example Usage

Create a VM running a Postgres instance

1
from sqlalchemy import text
2
import launchflow as lf
3

4
postgres = lf.aws.EC2Postgres("my-postgres-db")
5
engine = postgres.sqlalchemy_engine()
6

7
with engine.connect() as connection:
8
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)
content_copy

Create a VM running a Redis instance

1
import launchflow as lf
2

3
redis_vm = lf.aws.EC2Redis("my-redis-instance")
4

5
# Set a key-value pair
6
client = redis_vm.redis()
7
client.set("my-key", "my-value")
8

9
# Async compatible
10
async_client = await redis_vm.redis_async()
11
await async_client.set("my-key", "my-value")
content_copy

EC2

An EC2 instance running a Docker container.

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 ECS VM with a provided Docker image
4
ec2_instance = lf.aws.EC2("my-instance", vm_config=lf.aws.ec2.VMConfig(
5
    additional_outputs={"my_output": "my_value"},
6
    docker_cfg=lf.aws.ec2.DockerConfig(
7
        image="my-docker-image",
8
        args=[],
9
        environment_variables={"MY_ENV_VAR": "my_value"},
10
    ),
11
    firewall_cfg=lf.aws.ec2.FirewallConfig(expose_ports=[80]),
12
))
content_copy

initialization

Create a new EC2 resource.

Args:

• name (str): The name of the resource. This must be globally unique.
• vm_config (VMConfig): The configuration for the VM.
  • instance_type (str): The type of machine to use. Defaults to t3.micro for development environments and t3.medium for production environments.
  • additional_outputs (dict): Additional outputs to be returned by the resource.
  • docker_cfg (DockerConfig): The configuration for the Docker container.
    • image (str): The Docker image to run.
    • args (List[str]): The arguments to pass to the Docker container.
    • environment_variables (dict): Environment variables to set in the Docker container.
  • firewall_cfg (FirewallConfig): The configuration for the firewall rules.
    • expose_ports (List[int]): The ports to expose in the firewall.
  • disk_size_gb (int): The size of the disk in GB. Defaults to 8.

ssh

1
EC2.ssh()
content_copy

Open an SSH session to the VM.

Example usage:

1
import launchflow as lf
2

3
vm = lf.aws.EC2("my-vm")
4
vm.ssh()
content_copy

EC2Postgres

An EC2 instance running Postgres on Docker.

Example usage

1
from sqlalchemy import text
2
import launchflow as lf
3

4
postgres = lf.aws.EC2Postgres("my-postgres-db")
5
engine = postgres.sqlalchemy_engine()
6

7
with engine.connect() as connection:
8
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)
content_copy

initialization

Create a new EC2Postgres resource.

Args:

• name (str): The name of the Postgres VM resource. This must be globally unique.
• password (Optional[str]): The password for the Postgres DB. If not provided, a random password will be generated.
• instance_type (Optional[str): The type of machine to use. Defaults to t3.micro for development environments and t3.medium for production environments.
• disk_size_gb (Optional[str]): The size of the disk in GB. Defaults to 8.

query

1
EC2Postgres.query(query: str)
content_copy

Executes a SQL query on the Postgres instance running on the EC2 VM.

Args:

• query: The SQL query to execute.

Returns:

• The result of the query.

Example usage:

1
import launchflow as lf
2

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

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

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

django_settings

1
EC2Postgres.django_settings()
content_copy

Returns a Django settings dictionary for connecting to the Postgres instance running on EC2.

Returns:

• A dictionary of Django settings for connecting to the Postgres instance.

Example usage:

1
import launchflow as lf
2

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

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

sqlalchemy_engine_options

1
EC2Postgres.sqlalchemy_engine_options()
content_copy

Returns SQLAlchemy engine options for connecting to the Postgres instance running on EC2.

Returns:

• A dictionary of SQLAlchemy engine options.

sqlalchemy_async_engine_options

1
async EC2Postgres.sqlalchemy_async_engine_options()
content_copy

Returns async SQLAlchemy engine options for connecting to the Postgres instance running on EC2.

Returns:

• A dictionary of async SQLAlchemy engine options.

sqlalchemy_engine

1
EC2Postgres.sqlalchemy_engine(**engine_kwargs)
content_copy

Returns a SQLAlchemy engine for connecting to a postgres instance hosted on EC2.

Args:

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

Returns:

• A SQLAlchemy engine.

sqlalchemy_async_engine

1
async EC2Postgres.sqlalchemy_async_engine(**engine_kwargs)
content_copy

Returns an async SQLAlchemy engine for connecting to a postgres instance hosted on EC2.

Args:

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

Returns:

• An async SQLAlchemy engine.

EC2Redis

An EC2 instance running Redis on Docker.

Example usage

1
import launchflow as lf
2

3
redis_vm = lf.aws.EC2Redis("my-redis-instance")
4

5
# Set a key-value pair
6
client = redis_vm.redis()
7
client.set("my-key", "my-value")
8

9
# Async compatible
10
async_client = await redis_vm.redis_async()
11
await async_client.set("my-key", "my-value")
content_copy

initialization

Create a new EC2Redis resource.

Args:

• name (str): The name of the Redis VM resource. This must be globally unique.
• password (Optional[str]): The password for the Redis DB. If not provided, a random password will be generated.
• instance_type (Optional[str]): The type of machine to use. Defaults to t3.micro for development environments and t3.medium for production environments.
• disk_size_gb (Optional[int]): The size of the disk in GB. Defaults to 8.

django_settings

1
EC2Redis.django_settings()
content_copy

Returns a Django settings dictionary for connecting to the Redis instance running on EC2.

Returns:

• A dictionary of Django settings for connecting to the Redis instance.

Example usage:

1
import launchflow as lf
2

3
redis_vm = lf.aws.EC2Redis("my-redis-vm")
4

5
# settings.py
6
CACHES = {
7
    # Connect Django's cache backend to the Redis instance running on EC2
8
    "default": redis_vm.django_settings(),
9
}
content_copy

redis

1
EC2Redis.redis(*, decode_responses: bool = True)
content_copy

Get a Generic Redis Client object from the redis-py library.

Args:

• decode_responses (bool): Whether to decode responses from the Redis server. Defaults to True.

Returns:

• The Generic Redis Client from the redis-py library.

Example usage:

1
import launchflow as lf
2

3
redis_vm = lf.aws.EC2Redis("my-redis-instance")
4
client = redis_vm.redis()
5
client.set("my-key", "my-value")
content_copy

redis_async

1
async EC2Redis.redis_async(*, decode_responses: bool = True)
content_copy

Get an Async Redis Client object from the redis-py library.

Args:

• decode_responses (bool): Whether to decode responses from the Redis server. Defaults to True.

Returns:

• The Async Redis Client object from the redis-py library.

Example usage:

1
import launchflow as lf
2

3
redis_vm = lf.aws.EC2Redis("my-redis-instance")
4
async_client = await redis_vm.redis_async()
5
await async_client.set("my-key", "my-value")
content_copy

EC2SimpleServer

An EC2 instance running a simple server on Docker.

initialization

Create a new EC2SimpleServer resource.

Args:

• name (str): The name of the Redis VM resource. This must be globally unique.
• instance_type: The type of machine to use. Defaults to t3.micro for development environments and t3.medium for production environments.
• disk_size_gb: The size of the disk in GB. Defaults to 8.

EC2MySQL

An EC2 instance running MySQL on Docker.

Example usage

1
from sqlalchemy import text
2
import launchflow as lf
3

4
mysql = lf.aws.EC2MySQL("my-mysql-db")
5
engine = mysql.sqlalchemy_engine()
6

7
with engine.connect() as connection:
8
    print(connection.execute(text("SELECT 1")).fetchone())  # prints (1,)
content_copy

initialization

Create a new EC2MySQL resource.

Args:

• name (str): The name of the MySQL VM resource. This must be globally unique.
• password (Optional[str]): The password for the MySQL DB. If not provided, a random password will be generated.
• instance_type (Optional[str]): The type of machine to use. Defaults to t3.micro for development environments and t3.medium for production environments.
• disk_size_gb (Optional[str]): The size of the disk in GB. Defaults to 8.

query

1
EC2MySQL.query(query: str)
content_copy

Executes a SQL query on the MySQL instance running on the EC2 VM.

Args:

• query: The SQL query to execute.

Returns:

• The result of the query.

Example usage:

1
import launchflow as lf
2

3
mysql = lf.aws.EC2MySQL("my-mysql-db")
4

5
# Executes a query on the MySQL instance running on the EC2 VM
6
mysql.query("SELECT 1")
content_copy

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

django_settings

1
EC2MySQL.django_settings()
content_copy

Returns a Django settings dictionary for connecting to the MySQL instance running on EC2.

Returns:

• A dictionary of Django settings for connecting to the MySQL instance.

Example usage:

1
import launchflow as lf
2

3
mysql = lf.aws.EC2MySQL("my-mysql-db")
4

5
# settings.py
6
DATABASES = {
7
    # Connect Django's ORM to the MySQL instance running on EC2
8
    "default": mysql.django_settings(),
9
}
content_copy

sqlalchemy_engine_options

1
EC2MySQL.sqlalchemy_engine_options()
content_copy

Returns SQLAlchemy engine options for connecting to the MySQL instance running on EC2.

Returns:

• A dictionary of SQLAlchemy engine options.

sqlalchemy_async_engine_options

1
async EC2MySQL.sqlalchemy_async_engine_options()
content_copy

Returns async SQLAlchemy engine options for connecting to the MySQL instance running on EC2.

Returns:

• A dictionary of async SQLAlchemy engine options.

sqlalchemy_engine

1
EC2MySQL.sqlalchemy_engine(**engine_kwargs)
content_copy

Returns a SQLAlchemy engine for connecting to a MySQL instance hosted on EC2.

Args:

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

Returns:

• A SQLAlchemy engine.

sqlalchemy_async_engine

1
async EC2MySQL.sqlalchemy_async_engine(**engine_kwargs)
content_copy

Returns an async SQLAlchemy engine for connecting to a MySQL instance hosted on EC2.

Args:

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

Returns:

• An async SQLAlchemy engine.