FastAPI-Crons

Installation

FastAPI-Crons requires Python 3.10+ and can be installed using pip or your favorite package manager.

pip install fastapi-crons

poetry add fastapi-crons

conda install -c conda-forge fastapi-crons

Quick Start

Get up and running with FastAPI-Crons in under 5 minutes.

from fastapi import FastAPI
from fastapi_crons import Crons, get_cron_router

app = FastAPI(title="My App with Cron Jobs")
crons = Crons(app)

# Add cron management endpoints
app.include_router(get_cron_router(), prefix="/api")

@crons.cron("*/5 * * * *", name="cleanup")
async def cleanup_task():
    # Runs every 5 minutes
    print("🧹 Cleaning up temporary files...")
    return "Cleanup completed"

@crons.cron("0 0 * * *", name="daily_report")
def generate_daily_report():
    # Runs at midnight every day
    print("📊 Generating daily report...")
    return "Report generated"

Cron Expressions

FastAPI-Crons uses standard cron expressions to define when jobs should run. Master the syntax to schedule jobs precisely.

Job Definition

Learn different ways to define cron jobs in FastAPI-Crons, from simple decorators to advanced configurations.

@crons.cron("0 */2 * * *", name="data_sync", tags=["sync", "data"])
async def sync_data():
    # Runs every 2 hours
    print("Syncing data...")
    return "Data synced successfully"

from fastapi_crons import CronJob

def backup_database():
    print("Backing up database...")

job = CronJob(
    func=backup_database,
    expr="0 2 * * *",
    name="database_backup",
    tags=["backup", "database"]
)

crons.jobs.append(job)

Scheduler

The Crons class is the heart of FastAPI-Crons, managing job execution, state, and lifecycle.

State Management

FastAPI-Crons provides robust state management to track job execution history and status across restarts.

from fastapi_crons import Crons
from fastapi_crons.state import SQLiteStateBackend

state = SQLiteStateBackend(db_path="jobs.db")
crons = Crons(state_backend=state)

import redis.asyncio as redis
from fastapi_crons.state import RedisStateBackend

redis_client = redis.from_url("redis://localhost:6379")
state = RedisStateBackend(redis_client)
crons = Crons(state_backend=state)

Hooks System

Hooks provide powerful extension points to execute custom logic at different stages of job execution.

def log_job_start(job_name: str, context: dict):
    print(f"🚀 Starting job: {job_name}")
    print(f"📅 Scheduled: {context['scheduled_time']}")

crons.add_before_run_hook(log_job_start)

async def notify_success(job_name: str, context: dict):
    duration = context['duration']
    print(f"✅ {job_name} completed in {duration:.2f}s")
    
    # Send notification
    await send_slack_message(f"Job {job_name} succeeded")

crons.add_after_run_hook(notify_success)

async def handle_error(job_name: str, context: dict):
    error = context['error']
    print(f"❌ {job_name} failed: {error}")
    
    # Send alert
    await send_error_alert(job_name, error)

crons.add_on_error_hook(handle_error)

Job Tags

Organize and categorize your jobs with tags for better management and filtering.

@crons.cron("0 2 * * *", tags=["backup", "database", "critical"])
async def backup_database():
    # Critical database backup
    pass

@crons.cron("*/15 * * * *", tags=["cleanup", "maintenance"])
def cleanup_temp_files():
    # Regular maintenance task
    pass

@crons.cron("0 9 * * 1", tags=["reporting", "analytics"])
def weekly_report():
    # Weekly analytics report
    pass

# Hook that only runs for critical jobs
def critical_job_monitor(job_name: str, context: dict):
    if "critical" in context["tags"]:
        # Send immediate notification for critical jobs
        send_priority_alert(job_name, context)

# Apply different monitoring for different job types
def tag_based_monitoring(job_name: str, context: dict):
    tags = context["tags"]
    
    if "backup" in tags:
        log_backup_metrics(job_name, context)
    
    if "reporting" in tags:
        track_report_generation(job_name, context)

Distributed Locking

Prevent job overlap across multiple instances with Redis-based distributed locking.

from fastapi_crons import Crons, CronConfig
from fastapi_crons.locking import DistributedLockManager, RedisLockBackend
import redis.asyncio as redis

# Configure Redis connection
redis_client = redis.from_url("redis://localhost:6379")

# Setup distributed locking
config = CronConfig(enable_distributed_locking=True)
lock_backend = RedisLockBackend(redis_client)
lock_manager = DistributedLockManager(lock_backend, config)

crons = Crons(lock_manager=lock_manager, config=config)

Thread Safety

FastAPI-Crons is designed with thread safety in mind, ensuring reliable operation in concurrent environments.

Manual Execution

Trigger jobs manually for testing, debugging, or immediate execution outside their scheduled times.

curl -X POST "http://localhost:8000/api/crons/backup_database/run" \
     -H "Content-Type: application/json" \
     -d '{"force": false}'

{
  "status": "success",
  "message": "Job 'backup_database' executed successfully",
  "execution_time": 2.34,
  "instance_id": "worker-001",
  "timestamp": "2025-01-09T11:47:21Z"
}

from fastapi_crons import Crons
import asyncio

async def trigger_job_manually(job_name: str):
    crons = Crons()
    job = crons.get_job(job_name)
    
    if not job:
        raise ValueError(f"Job '{job_name}' not found")
    
    # Execute the job function directly
    if asyncio.iscoroutinefunction(job.func):
        result = await job.func()
    else:
        result = await asyncio.to_thread(job.func)
    
    return result

# Usage
result = await trigger_job_manually("backup_database")

CLI Commands

Comprehensive command-line interface for managing, monitoring, and debugging your cron jobs.

Crons Class

The main scheduler class that manages all cron jobs in your application.

class Crons:
    def __init__(
        self,
        app: FastAPI = None,
        state_backend: StateBackend = None,
        lock_manager: DistributedLockManager = None,
        config: CronConfig = None
    ):

@crons.cron("0 * * * *", name="hourly")
def my_job():
    pass

jobs = crons.get_jobs()
for job in jobs:
    print(job.name)

job = crons.get_job("backup")
if job:
    print(job.next_run)

crons.add_before_run_hook(log_start)

await crons.start()

await crons.stop()

Frequently Asked Questions