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"}

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

const dailyCleanup = defineWorkflow<SchedulePayload, unknown, CleanupResult>(
  {
    id: 'daily_cleanup',
    schedule: '0 3 * * *',
  },
  async (ctx, payload) => {
    // Runs daily at 3:00 AM UTC
    const cleanupResult = await ctx.step.run('cleanup_old_records', () => cleanupOldRecords());
    const tempResult = await ctx.step.run('cleanup_temp_files', () => cleanupTempFiles());
    return { timestamp: payload.timestamp, recordsCleaned: cleanupResult.count, filesCleaned: tempResult.count };
  },
);

@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}

const morningReport = defineWorkflow<SchedulePayload, unknown, ReportResult>(
  {
    id: 'morning_report',
    schedule: { cron: '0 8 * * 1-5', timezone: 'America/New_York' },
  },
  async (ctx, payload) => {
    // Runs at 8:00 AM Eastern, Monday-Friday
    const metrics = await ctx.step.run('gather_metrics', () => gatherDailyMetrics());
    const report = await ctx.step.run('generate_report', () => generateReport(metrics));
    await ctx.step.run('send_report', () => sendReport(report));
    return { timestamp: payload.timestamp, reportId: report.id, metricsCount: Object.keys(metrics).length };
  },
);

@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"]}

const hourlySync = defineWorkflow<SchedulePayload, unknown, SyncResult>(
  {
    id: 'hourly_sync',
    schedule: '0 * * * *',
  },
  async (ctx, payload) => {
    // Runs at the start of every hour
    const syncResult = await ctx.step.run('sync_external_data', () => syncExternalData());
    return { timestamp: payload.timestamp, recordsSynced: syncResult.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}

const schedulableReminder = defineWorkflow<SchedulePayload, unknown, ReminderResult>(
  {
    id: 'schedulable_reminder',
    schedule: true, // No default cron, can be scheduled via API
  },
  async (ctx, payload) => {
    await ctx.step.run('send_reminder', () => sendReminder());
    return { timestamp: payload.timestamp, message: 'Scheduled reminder!', 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"],
    )

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

const polos = new Polos();

// Create a schedule for a specific user
const scheduleId = await polos.schedules.create(
  'schedulable_reminder',
  '0 8 * * *',
  'America/New_York',
  'user-alice', // Unique key for this schedule
);

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

for (const user of users) {
  await polos.schedules.create(
    'schedulable_reminder',
    user.cron,
    user.tz,
    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

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