launchflow.gcp.compute_engine

ComputeEngine

1
class ComputeEngine(Resource[T])

__init__

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

Generic compute engine resource.

Arguments:

  • name (str): The name of the resource. This must be globally unique.
  • vm_config (VMConfig): The configuration for the VM.

ComputeEnginePostgres

1
class ComputeEnginePostgres(ComputeEngine[ComputeEnginePostgresConnectionInfo]
2
)

__init__

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

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

Arguments:

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

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

sqlalchemy_engine

1
def sqlalchemy_engine(**engine_kwargs)

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

sqlalchemy_async_engine

1
async def sqlalchemy_async_engine(**engine_kwargs)

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

ComputeEngineRedis

1
class ComputeEngineRedis(Resource[ComputeEngineRedisConnectionInfo])

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

Attributes:

  • name (str): The name of the Redis VM resource. This must be globally unique.

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

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: