Migration Guide: Decorator-Based to Dynamic Workflow Scheduling#

@DBOS.scheduled("*/5 * * * * *")
@DBOS.workflow()
def my_workflow(scheduled: datetime, actual: datetime) -> None:
    # The second parameter is the actual execution time
    logger.info(f"Scheduled for {scheduled}, actually ran at {actual}")
    # ... workflow logic

@DBOS.workflow()
def my_workflow(scheduled_time: datetime, context: Any) -> None:
    # The second parameter is now a custom context object
    # If you need actual time, capture it manually
    actual_time = datetime.now(timezone.utc)

    logger.info(f"Scheduled for {scheduled_time}, context: {context}")
    logger.info(f"Actually ran at {actual_time}")
    # ... workflow logic

# Create schedule at application startup or via API endpoint
DBOS.create_schedule(
    schedule_name="my-schedule",
    workflow_fn=my_workflow,
    schedule="*/5 * * * * *", # Note: seconds are supported
    context={"env": "production", "version": "1.0"}, # Optional: any JSON-serializable value
    automatic_backfill=False, # Optional: enable automatic backfill on startup
    cron_timezone=None, # Optional: IANA timezone (e.g., "America/New_York")
    queue_name=None, # Optional: name of a declared queue for concurrency management
)

DBOS.delete_schedule("my-schedule")

# List all schedules
schedules = DBOS.list_schedules()
for schedule in schedules:
    print(f"{schedule.schedule_name}: {schedule.schedule}")

# Pause a schedule (stops execution but keeps config)
DBOS.pause_schedule("my-schedule")

# Resume a paused schedule
DBOS.resume_schedule("my-schedule")

# Trigger immediate execution (ignores cron schedule)
DBOS.trigger_schedule("my-schedule")

# Delete a schedule permanently
DBOS.delete_schedule("my-schedule")

# Backfill missed executions over a date range
DBOS.backfill_schedule(
    schedule_name="my-schedule",
    start_time=datetime(2026, 1, 1),
    end_time=datetime(2026, 1, 31),
)

# Atomic update of multiple schedules
# Note: context is optional; when omitted, the workflow receives None
DBOS.apply_schedules([
    ScheduleInput(
        schedule_name="schedule-1",
        workflow_fn=workflow_a,
        schedule="0 * * * *",
        context={"id": 1}, # Optional: any JSON-serializable value
        automatic_backfill=False,
        cron_timezone=None,
        queue_name=None, # Optional: enqueue to a declared queue
    ),
    ScheduleInput(
        schedule_name="schedule-2",
        workflow_fn=workflow_b,
        schedule="0 */2 * * *",
        # context omitted - workflow will receive None
        automatic_backfill=True,
        cron_timezone="America/New_York",
        queue_name="high-priority-queue", # Must be declared before use
    ),
])

from dbos import DBOS, Queue

# Declare a queue for rate-limited workflows
rate_limited_queue = Queue("rate-limited", concurrency=5)

# Schedule workflows to use the queue
DBOS.create_schedule(
    schedule_name="rate-limited-task",
    workflow_fn=my_workflow,
    schedule="*/5 * * * * *",
    context={"task": "batch-processing"},
    queue_name="rate-limited", # Must be declared before use
)

# List all application versions (newest first)
versions = DBOS.list_application_versions()
for version in versions:
    print(f"{version['version_name']}: {version['version_timestamp']}")

# Get the latest application version
latest = DBOS.get_latest_application_version()
print(f"Latest version: {latest['version_name']}")

# Set a specific version as latest (updates its timestamp)
DBOS.set_latest_application_version("v1.2.3")

from datetime import datetime
from dbos import DBOS

@DBOS.scheduled("*/5 * * * * *")
@DBOS.workflow()
def data_sync_workflow(scheduled: datetime, actual: datetime) -> None:
    """Syncs data every 5 seconds"""
    print(f"Running data sync at {actual}")
    # Sync logic here

@DBOS.scheduled("0 0 * * *")
@DBOS.workflow()
def daily_report(scheduled: datetime, actual: datetime) -> None:
    """Generates daily report at midnight"""
    print(f"Generating report for {scheduled.date()}")
    # Report generation logic

from datetime import datetime
from typing import Any
from dbos import DBOS

@DBOS.workflow()
def data_sync_workflow(scheduled_time: datetime, context: Any) -> None:
    """Syncs data every 5 seconds"""
    actual_time = datetime.now(timezone.utc)
    # context is optional; handle None if not provided
    env = context.get("env", "unknown") if context else "unknown"

    print(f"Running data sync in {env} environment at {actual_time}")
    # Sync logic here

@DBOS.workflow()
def daily_report(scheduled_time: datetime, context: Any) -> None:
    """Generates daily report at midnight"""
    # context is optional; handle None if not provided
    report_type = context.get("report_type", "standard") if context else "standard"

    print(f"Generating {report_type} report for {scheduled_time.date()}")
    # Report generation logic

# Create schedules at application startup
def initialize_schedules():
    DBOS.create_schedule(
        schedule_name="data-sync",
        workflow_fn=data_sync_workflow,
        schedule="*/5 * * * * *",
        context={"env": "production", "priority": "high"},
    )

    DBOS.create_schedule(
        schedule_name="daily-report",
        workflow_fn=daily_report,
        schedule="0 0 * * *",
        context={"report_type": "comprehensive", "email": True},
        cron_timezone="America/New_York", # 9 AM Eastern = midnight local
        automatic_backfill=True, # Ensure no reports are missed
    )

# Later: Pause for maintenance
DBOS.pause_schedule("data-sync")

# Resume after maintenance
DBOS.resume_schedule("data-sync")

# Backfill missed reports
DBOS.backfill_schedule(
    schedule_name="daily-report",
    start_time=datetime(2026, 2, 1),
    end_time=datetime(2026, 2, 10),
)