Scheduled Workflows

from polos import workflow, WorkflowContext, SchedulePayload

@workflow(id="daily_cleanup", schedule="0 3 * * *")
async def daily_cleanup(ctx: WorkflowContext, payload: SchedulePayload):
    """Runs daily at 3:00 AM UTC."""
    await ctx.step.run("cleanup_records", cleanup_old_records)
    await ctx.step.run("cleanup_files", cleanup_temp_files)
    return {"timestamp": payload.timestamp, "status": "completed"}

@workflow(
    id="morning_report",
    schedule={"cron": "0 8 * * 1-5", "timezone": "America/New_York"},
)
async def morning_report(ctx: WorkflowContext, payload: SchedulePayload):
    """Runs at 8:00 AM Eastern, Monday-Friday."""
    metrics = await ctx.step.run("gather", gather_metrics)
    await ctx.step.run("send", send_report, metrics)
    return {"timestamp": payload.timestamp}

@workflow(id="hourly_sync", schedule="0 * * * *")
async def hourly_sync(ctx: WorkflowContext, payload: SchedulePayload):
    """Runs at the start of every hour."""
    result = await ctx.step.run("sync", sync_external_data)
    return {"synced": result["count"]}

@workflow(id="reminder", schedule=True)
async def reminder(ctx: WorkflowContext, payload: SchedulePayload):
    """Can be scheduled dynamically via API."""
    await ctx.step.run("send", send_reminder)
    return {"sent": True}

from polos import Polos, schedules

polos = Polos()

# Create a schedule for a specific user
schedule_id = await schedules.create(
    polos=polos,
    workflow="reminder",
    cron="0 8 * * *",
    timezone="America/New_York",
    key="user-alice",  # Unique key for this schedule
)

# Create different schedules for different users
users = [
    {"id": "user-alice", "cron": "0 8 * * *", "tz": "America/New_York"},
    {"id": "user-bob", "cron": "0 9 * * *", "tz": "Europe/London"},
]

for user in users:
    await schedules.create(
        polos=polos,
        workflow="reminder",
        cron=user["cron"],
        timezone=user["tz"],
        key=user["id"],
    )

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