Workflow Schedules (#581)
Add dynamic workflow scheduling. Schedules are persisted in the database
and can be created, queried, or updated dynamically from workflows or
from the client.
API:
```python
@DBOS.workflow()
def my_periodic_task(scheduled_time: datetime, context: Any):
DBOS.logger.info(f"Running task scheduled for {scheduled_time} with context {context}")
# Create a new schedule
DBOS.create_schedule(
schedule_name="my-task-schedule", # The schedule name is a unique identifier of the schedule
workflow_fn=my_periodic_task,
schedule="*/5 * * * *", # Every 5 minutes
context="my context", # The context is passed into every iteration of the workflow
)
# Atomically apply a set of schedule changes
DBOS.apply_schedules([
{"schedule_name": "schedule-a", "workflow_fn": workflow_a, "schedule": "*/10 * * * *"}, # Every 10 minutes
{"schedule_name": "schedule-b", "workflow_fn": workflow_b, "schedule": "0 0 * * *"}, # Every day at midnight
])
# Pause a schedule so it stops firing
DBOS.pause_schedule("my-task-schedule")
# Resume a paused schedule
DBOS.resume_schedule("my-task-schedule")
# Delete a schedule
DBOS.delete_schedule("my-task-schedule")
# List all active schedules
schedules = DBOS.list_schedules(status="ACTIVE")
# Get a specific schedule by name
schedule = DBOS.get_schedule("my-task-schedule")
# Backfill a schedule
handles = DBOS.backfill_schedule(
"my-task-schedule",
start=datetime(2025, 1, 1, tzinfo=timezone.utc),
end=datetime(2025, 1, 2, tzinfo=timezone.utc),
)
# Trigger a schedule immediately
handle = DBOS.trigger_schedule("my-task-schedule")
``` P
Peter Kraft committed
ea2497bec187f897a86f8f7c8a37a9e00a95a13a
Parent: 209d433
Committed by GitHub <noreply@github.com>
on 2/12/2026, 10:19:42 PM