Cron jobs that live inside your FastAPI app
Define scheduled work with a decorator, next to the code it belongs to. Async native, with hooks, retries, timeouts, distributed locking, persistent state, a CLI and a monitoring dashboard.
from fastapi import FastAPI
from fastapi_crons import Crons
app = FastAPI()
crons = Crons(app)
@crons.cron("*/5 * * * *", name="cleanup")
async def cleanup_task():
"""Runs every 5 minutes."""
await purge_expired_sessions()
return "cleanup complete"Why FastAPI-Crons
Scheduled work usually ends up in a separate worker process with its own deployment, its own configuration and its own copy of your application code. FastAPI-Crons keeps jobs in the app they belong to: the scheduler starts and stops with the FastAPI lifespan, and jobs share your existing dependencies and settings.
Decorator based
Register a job with @crons.cron("*/5 * * * *"). No separate schedule file to keep in sync.
Sync and async
Async functions are awaited on the event loop; sync functions run in a thread so they never block it.
Hooks
Run logic before, after or on failure — for logging, metrics, alerting or webhooks.
Retries and timeouts
Per-job retry counts with exponential backoff and jitter, plus a hard timeout to stop runaway jobs.
Distributed locking
Redis, SQL or PostgreSQL advisory locks stop the same job running twice across replicas.
Persistent state
Last-run times and statuses survive restarts, via SQLite, Redis, SQLAlchemy or SQLModel.
Install
The base install has everything you need to schedule jobs and persist state to SQLite.
pip install fastapi-cronsOptional features are separate extras so the base install stays small — see Installation for the full list.
How it works
Crons(app) registers startup and shutdown handlers on your FastAPI application. On startup the scheduler loads registered jobs, restores their state from the configured backend and starts one loop per job. Each loop sleeps until the next time its cron expression matches, then runs the job — acquiring a distributed lock first if one is configured.
Jobs run in your app process
If you run several replicas behind a load balancer, every replica schedules every job. Enable distributed locking so only one replica actually executes each run.