FastScheduler

The FastScheduler constructor accepts several configuration parameters to control persistence, performance, and history:

• state_file: Path to the JSON persistence file (default: fastscheduler_state.json).
• storage: Backend type, either "json" (default) or "sqlmodel".
• database_url: Connection string for the sqlmodel backend.
• quiet: If True, suppresses log messages.
• auto_start: If True, starts the scheduler immediately.
• max_history: Maximum number of execution history entries to keep (default: 10000).
• max_workers: Number of concurrent job threads (default: 10).
• history_retention_days: Number of days to keep history before deletion (default: 7).
• max_dead_letters: Maximum failed jobs to keep in the Dead Letter Queue (default: 500).

scheduler = FastScheduler(
    state_file="scheduler.json",    # Persistence file for JSON backend
    storage="json",                 # Storage backend: "json" or "sqlmodel"
    database_url=None,              # Database URL for sqlmodel backend
    quiet=True,                     # Suppress log messages
    auto_start=False,               # Start immediately
    max_history=5000,               # Max history entries to keep
    max_workers=20,                 # Concurrent job threads
    history_retention_days=8,       # Delete history older than X days
    max_dead_letters=500,           # Max failed jobs in dead letter queue
)

The FastScheduler class can be initialized with several storage options to persist job state, history, and statistics.

• JSON (Default): Uses a local JSON file. Requires state_file parameter.
• SQLModel: Uses a relational database. Requires storage="sqlmodel" and a database_url (e.g., SQLite, PostgreSQL, or MySQL).
• Custom: Pass an instance of a class implementing the StorageBackend interface.

Configuration options include max_workers (thread pool size), max_history (limit on history entries), and history_retention_days (age limit for history).

# JSON storage (default)
scheduler = FastScheduler(state_file="scheduler.json")

# SQLite database via SQLModel
scheduler = FastScheduler(
    storage="sqlmodel",
    database_url="sqlite:///scheduler.db"
)

# PostgreSQL database via SQLModel
scheduler = FastScheduler(
    storage="sqlmodel",
    database_url="postgresql://user:pass@localhost/mydb"
)

# Custom storage backend
scheduler = FastScheduler(storage=MyCustomStorageBackend())

To add a web-based monitoring dashboard and management API to your FastAPI application, use create_scheduler_routes. This function returns an APIRouter containing several endpoints for viewing statistics, managing jobs (pause, resume, cancel, run now), and inspecting job history or dead letters.

Installation Requirement: You must install the FastAPI extra for this integration to work:

pip install fastscheduler[fastapi]

Default Configuration:

• The default URL prefix is /scheduler.
• The default OpenAPI tag is scheduler.
• You can override these by passing prefix and tags as keyword arguments to create_scheduler_routes.

from fastapi import FastAPI
from fastscheduler import FastScheduler
from fastscheduler.fastapi_integration import create_scheduler_routes

app = FastAPI()
scheduler = FastScheduler()

# Include the scheduler routes in your FastAPI app
app.include_router(create_scheduler_routes(scheduler))

scheduler.start()

This example demonstrates how to define various types of jobs (interval, async with timezone, cron with retries, and weekly) and manage the scheduler lifecycle using start() and stop().

import asyncio
import time
from fastscheduler import FastScheduler

scheduler = FastScheduler(quiet=True)

# Simple interval job
@scheduler.every(10).seconds
def heartbeat():
    print(f"[{time.strftime('%H:%M:%S')}] ❤️ Heartbeat")

# Async job with timezone
@scheduler.daily.at("09:00", tz="America/New_York").timeout(60)
async def morning_report():
    print("Generating report...")
    await asyncio.sleep(5)
    print("Report sent!")

# Cron job with retries
@scheduler.cron("*/5 * * * *").retries(3)
def check_api():
    print("Checking API health")

# Weekly job
@scheduler.weekly.monday.at("10:00")
def weekly_standup():
    print("Time for standup!")

# Start scheduler
scheduler.start()

try:
    while True:
        time.sleep(60)
        scheduler.print_status()
except KeyboardInterrupt:
    scheduler.stop()