State Persistence

from polos import WorkflowState

class ShoppingCartState(WorkflowState):
    items: list[dict] = []
    total: float = 0.0

from polos import workflow, WorkflowContext

@workflow(id="shopping_cart", state_schema=ShoppingCartState)
async def shopping_cart(ctx: WorkflowContext, payload: CartPayload):
    if payload.action == "add" and payload.item:
        ctx.state.items.append(payload.item.model_dump())
        ctx.state.total += payload.item.price * payload.item.quantity

    elif payload.action == "remove" and payload.item_id:
        for i, item in enumerate(ctx.state.items):
            if item.get("id") == payload.item_id:
                ctx.state.total -= item["price"] * item["quantity"]
                ctx.state.items.pop(i)
                break

    elif payload.action == "clear":
        ctx.state.items = []
        ctx.state.total = 0.0

    return {"items": ctx.state.items, "total": ctx.state.total}

# Start workflow with initial state
handle = await polos.invoke(
    "shopping_cart",
    payload={"action": "add", "item": {...}},
    initial_state={"items": [], "total": 0.0},
)

git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/10-state-persistence
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
uv sync
python main.py