launchflow.gcp.compute_engine

ComputeEngine

1
class ComputeEngine(GCPResource[T])
content_copy

A Compute Engine VM running a Docker container.

Example usage:

1
import launchflow as lf
2

3
compute_engine = lf.gcp.ComputeEngine("my-compute-engine", vm_config=lf.gcp.compute_engine.VMConfig(
4
    additional_outputs={"my_output": "my_value"},
5
    docker_cfg=lf.gcp.compute_engine.DockerConfig(
6
        image="my-docker-image",
7
        args=[],
8
        environment_variables={"MY_ENV_VAR": "my_value"},
9
    ),
10
    firewall_cfg=lf.gcp.compute_engine.FirewallConfig(expose_ports=[80]),
11
))
content_copy

__init__

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

Create a Compute Engine 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.

ComputeEnginePostgres

1
class ComputeEnginePostgres(ComputeEngine[ComputeEnginePostgresConnectionInfo]
2
                            )
content_copy

A Postgres instance running on a VM in Google Compute Engine.

Example usage:

1
from sqlalchemy import text
2
import launchflow as lf
3

4
postgres_compute_engine = lf.gcp.ComputeEnginePostgres("ce-postgres-mn-test-2")
5
engine = postgres_compute_engine.sqlalchemy_engine()
6

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

__init__

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

Create a new Compute Engine Postgres 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)
content_copy

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

Args:

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

Example usage:

1
import launchflow as lf
2
db = lf.gcp.ComputeEnginePostgres("my-pg-db")
3
engine = db.sqlalchemy_engine()
content_copy

sqlalchemy_async_engine

1
async def sqlalchemy_async_engine(**engine_kwargs)
content_copy

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

Args:

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

Example usage:

1
import launchflow as lf
2
db = lf.gcp.ComputeEnginePostgres("my-pg-db")
3
engine = await db.sqlalchemy_async_engine()
content_copy

ComputeEngineRedis

1
class ComputeEngineRedis(ComputeEngine[ComputeEngineRedisConnectionInfo])
content_copy

A Redis resource running on a VM in Google Compute Engine.

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.

Example usage:

1
import launchflow as lf
2

3
redis = lf.gcp.ComputeEngineRedis("my-redis-instance")
4

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

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

redis

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

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

Returns:

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

redis_async

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

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

Returns:

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