waymark quickstart

run.py

# /// script
# requires-python = "==3.12"
# dependencies = ["waymark"]
#
# [tool.uv.sources]
# waymark = { path = "python", editable = true }
# ///

import asyncio
import os
import sys
from dataclasses import dataclass
from typing import Annotated

# import self as named module so workers can reload it
import run as _mod

# postgres or fall back to in-memory
os.environ["WAYMARK_DATABASE_URL"] = sys.argv[1]

from waymark import Depend, Workflow, action, workflow
from waymark.workflow import workflow_registry

workflow_registry._workflows.clear()


@dataclass
class User:
    id: str
    email: str
    active: bool


@dataclass
class EmailResult:
    to: str
    subject: str
    success: bool


async def get_mock_db():
    return {
        "user1": User(id="user1", email="alice@example.com", active=True),
        "user2": User(id="user2", email="bob@example.com", active=False),
        "user3": User(id="user3", email="carol@example.com", active=True),
    }


@action
async def fetch_users(
    user_ids: list[str],
    db: Annotated[dict, Depend(get_mock_db)],
) -> list[User]:
    return [db[uid] for uid in user_ids if uid in db]


@action
async def send_email(
    to: str,
    subject: str,
) -> EmailResult:
    return EmailResult(to=to, subject=subject, success=True)


@workflow
class WelcomeEmailWorkflow(Workflow):
    async def run(self, user_ids: list[str]) -> dict:
        users = await fetch_users(user_ids)
        active_users = [user for user in users if user.active]

        results = await asyncio.gather(
            *[send_email(to=user.email, subject="Welcome") for user in active_users],
            return_exceptions=True,
        )

        return {
            "total_users": len(users),
            "active_users": len(active_users),
            "emails_sent": len(results),
        }


async def main():
    result = await _mod.WelcomeEmailWorkflow().run(user_ids=["user1", "user2", "user3"])
    print(f"{result=}")


if __name__ == "__main__":
    asyncio.run(main())

run.sh

#!/bin/bash
set -euox pipefail

# clean
pkill -f waymark-bridge 2>/dev/null || true
pkill -f start-workers 2>/dev/null || true
docker rm -f tmp 2>/dev/null || true
sleep 1

# start ephemeral postgres
docker run -d --name tmp --rm \
    -e POSTGRES_USER=demo -e POSTGRES_DB=demo -e POSTGRES_PASSWORD=demo \
    -p 5432:5432 postgres:17-alpine >/dev/null
# wait until ready
until docker exec tmp pg_isready >/dev/null 2>&1; do sleep 1; done

# build
cargo build --bin boot-waymark-singleton --bin waymark-bridge --bin start-workers
uv sync --dev --directory python

# start worker daemon
DB_URL="postgresql://demo:demo@localhost:5432/demo"
PATH="$(pwd)/python/.venv/bin:$PATH" \
    PYTHONPATH="$(pwd)" \
    WAYMARK_DATABASE_URL="$DB_URL" \
    WAYMARK_USER_MODULE=run \
    ./target/debug/start-workers &
START_WORKERS_PID=$!
trap 'kill "$START_WORKERS_PID" 2>/dev/null || true' EXIT
# wait until ready
until docker exec tmp psql -U demo -d demo -c "SELECT 1 FROM queued_instances LIMIT 0" >/dev/null 2>&1; do sleep 1; done

# run
uv run run.py "$DB_URL"

# dump db
# docker exec tmp pg_dump -U demo --inserts demo > pg_out.txt

curl https://gist.githubusercontent.com/sueszli/70d1fc670e9e805eb1f6d42cc923d70d/raw/faf6f38ee80c1905d3850380eda2b2c809a4bcab/run.py > run.py

curl https://gist.githubusercontent.com/sueszli/70d1fc670e9e805eb1f6d42cc923d70d/raw/faf6f38ee80c1905d3850380eda2b2c809a4bcab/run.sh > run.sh

chmod +x ./run.sh && ./run.sh