Workflow Basics

from polos import workflow, WorkflowContext
from pydantic import BaseModel

class OrderPayload(BaseModel):
    order_id: str
    items: list[str]
    total: float

@workflow(id="process_order")
async def process_order(ctx: WorkflowContext, payload: OrderPayload):
    # Steps are automatically retried on failure
    await ctx.step.run("validate", validate_order, payload)
    await ctx.step.run("reserve", reserve_inventory, payload.items)
    await ctx.step.run("charge", process_payment, payload.total)

    # Deterministic UUID (same on replay)
    confirmation = await ctx.step.uuid("confirmation")

    return {"status": "completed", "confirmation": confirmation}

import { defineWorkflow } from '@polos/sdk';

interface OrderPayload {
  orderId: string;
  items: string[];
  total: number;
}

const processOrder = defineWorkflow<OrderPayload, unknown, Record<string, unknown>>(
  { id: 'process_order' },
  async (ctx, payload) => {
    // Steps are automatically retried on failure
    await ctx.step.run('validate', () => validateOrder(payload));
    await ctx.step.run('reserve', () => reserveInventory(payload.items));
    await ctx.step.run('charge', () => processPayment(payload.total));

    // Deterministic UUID (same on replay)
    const confirmation = await ctx.step.uuid('confirmation');

    return { status: 'completed', confirmation };
  },
);

await ctx.step.run(
    "unreliable_api",
    call_external_api,
    data,
    max_retries=5,
    base_delay=2.0,
    max_delay=30.0,
)

await ctx.step.run(
  'unreliable_api',
  () => callExternalApi(data),
  {
    maxRetries: 5,
    baseDelay: 2000,   // ms
    maxDelay: 30000,    // ms
  },
);

# These return the same value on replay
random_value = await ctx.step.random("coin_flip")
unique_id = await ctx.step.uuid("order_id")
timestamp = await ctx.step.now("created_at")

// These return the same value on replay
const randomValue = await ctx.step.random('coin_flip');
const uniqueId = await ctx.step.uuid('order_id');
const timestamp = await ctx.step.now('created_at');

await ctx.step.wait_for("cooldown", seconds=60)

await ctx.step.waitFor('cooldown', { seconds: 60 });

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

git clone https://github.com/polos-dev/polos.git
cd polos/typescript-examples/08-workflow-basics
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
npm install
npx tsx main.ts