Documentation Index
Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
Use this file to discover all available pages before exploring further.
Run workflows automatically on a schedule using cron expressions.
Daily schedule
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"}
Schedule with timezone
@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}
Hourly sync
@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"]}
Schedulable (no default schedule)
@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}
Create schedules dynamically
Use schedules.create() to create schedules at runtime. This is useful for per-user or per-entity schedules.
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"],
)
Cron syntax
┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sun=0)
│ │ │ │ │
* * * * *
0 3 * * * - Daily at 3:00 AM0 * * * * - Every hour*/15 * * * * - Every 15 minutes0 8 * * 1-5 - Weekdays at 8:00 AM
Run it
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
