LaunchFlow Logo

launchflow.aws.ec2

EC2

1
class EC2(AWSResource[T])

An EC2 instance running a Docker container.

Example usage:

1
import launchflow as lf

2


3
ec2_instance = lf.aws.EC2("my-instance", vm_config=lf.aws.ec2.VMConfig(

4
    additional_outputs={"my_output": "my_value"},

5
    docker_cfg=lf.aws.ec2.DockerConfig(

6
        image="my-docker-image",

7
        args=[],

8
        environment_variables={"MY_ENV_VAR": "my_value"},

9
    ),

10
    firewall_cfg=lf.aws.ec2.FirewallConfig(expose_ports=[80]),

11
))

__init__

1
def __init__(name: str, vm_config: VMConfig) -> None

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.
    • 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.

ssh

1
def ssh()

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()

EC2Postgres

1
class EC2Postgres(EC2[EC2PostgresConnectionInfo])

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,)

__init__

1
def __init__(name: str, *, password: Optional[str] = None) -> None

Create a new EC2Postgres resource.

Args:

  • name (str): The name of the Postgres VM resource. This must be globally unique.
  • password (str): The password for the Postgres DB. If not provided, a random password will be generated.

sqlalchemy_engine

1
def sqlalchemy_engine(**engine_kwargs)

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.

Example usage:

1
import launchflow as lf

2
db = lf.aws.EC2Postgres("my-pg-db")

3
engine = db.sqlalchemy_engine()

sqlalchemy_async_engine

1
async def sqlalchemy_async_engine(**engine_kwargs)

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.

Example usage:

1
import launchflow as lf

2
db = lf.aws.EC2Postgres("my-pg-db")

3
engine = await db.sqlalchemy_async_engine()

EC2Redis

1
class EC2Redis(EC2[EC2RedisConnectionInfo])

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")

__init__

1
def __init__(name: str, *, password: Optional[str] = None) -> None

Create a new EC2Redis resource.

Args:

  • name (str): The name of the Redis VM resource. This must be globally unique.
  • password (str): The password for the Redis DB. If not provided, a random password will be generated.

django_settings

1
def django_settings()

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

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
}

redis

1
def redis(*, decode_responses: bool = True)

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

Returns:

redis_async

1
async def redis_async(*, decode_responses: bool = True)

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

Returns: