Effortlessly schedule and manage your background tasks.

Built with the tools and technologies:

Welcome to the official guide for using fastapi_crons, a high-performance, developer-friendly cron scheduling extension for FastAPI. This library enables you to define, monitor, and control scheduled background jobs using simple decorators and provides CLI tools, web-based monitoring, and SQLite-based job tracking.

• Native integration with FastAPI using from fastapi import FastAPI, Crons
• Define cron jobs with decorators
• Async + sync job support
• SQLite job state persistence
• CLI for listing and managing jobs
• Automatic monitoring endpoint (/crons)
• Named jobs, tags, and metadata
• Easy to plug into any FastAPI project

pip install fastapi-crons

from fastapi import FastAPI from fastapi_crons import Crons, get_cron_router app = FastAPI() crons = Crons(app) app.include_router(get_cron_router()) @app.get("/") def root(): return {"message": "Hello from FastAPI"}

@crons.cron("*/5 * * * *", name="print_hello") def print_hello(): print("Hello! I run every 5 minutes.") @crons.cron("0 0 * * *", name="daily_task", tags=["rewards"]) async def run_daily_task(): # Distribute daily rewards or any async task await some_async_function()

┌───────────── minute (0 - 59) │ ┌───────────── hour (0 - 23) │ │ ┌───────────── day of the month (1 - 31) │ │ │ ┌───────────── month (1 - 12) │ │ │ │ ┌───────────── day of the week (0 - 6) (Sunday to Saturday) │ │ │ │ │ * * * * *

• * * * * *: Every minute
• */15 * * * *: Every 15 minutes
• 0 * * * *: Every hour
• 0 0 * * *: Every day at midnight
• 0 0 * * 0: Every Sunday at midnight

Once included, visit:

GET /crons 

You'll get a full list of jobs with:

• name
• expr (cron expression)
• tags
• last_run (from SQLite)
• next_run

We use SQLite (via aiosqlite) to keep a persistent record of when each job last ran. This allows observability and resilience during restarts.

For projects already using SQLAlchemy or SQLModel with PostgreSQL or MySQL, you can reuse your existing database connection instead of SQLite.

pip install fastapi-crons[sqlalchemy] # or pip install fastapi-crons[sqlmodel]

from sqlalchemy.ext.asyncio import create_async_engine from fastapi_crons import Crons from fastapi_crons.state.sqlalchemy import SQLAlchemyStateBackend engine = create_async_engine("postgresql+asyncpg://user:pass@host/db") backend = SQLAlchemyStateBackend(engine) crons = Crons(app, state_backend=backend)

Works with sync engines too:

from sqlalchemy import create_engine engine = create_engine("postgresql://user:pass@host/db") backend = SQLAlchemyStateBackend(engine)

# env.py from fastapi_crons.state.sqlalchemy import cron_metadata target_metadata = [Base.metadata, cron_metadata]

pip install fastapi-crons[sqlalchemy]

from fastapi_crons.locking.sqlalchemy import SQLAlchemyLockBackend from fastapi_crons.locking import DistributedLockManager from fastapi_crons import CronConfig engine = create_async_engine("postgresql+asyncpg://...") backend = SQLAlchemyLockBackend(engine) manager = DistributedLockManager(backend, CronConfig()) crons = Crons(app, lock_manager=manager)

For PostgreSQL, advisory locks are available as a lighter alternative (no table required):

from fastapi_crons.locking.sqlalchemy import PostgreSQLAdvisoryLockBackend backend = PostgreSQLAdvisoryLockBackend(engine)

CREATE TABLE IF NOT EXISTS job_state ( name TEXT PRIMARY KEY, last_run TEXT );

By default, job state is stored in a SQLite database named cron_state.db in the current directory. You can customize the database path:

from fastapi_crons import Crons, SQLiteStateBackend # Custom database path state_backend = SQLiteStateBackend(db_path="/path/to/my_crons.db") crons = Crons(state_backend=state_backend)

The scheduler supports both async and sync job functions Jobs can be:

• async def → run in asyncio loop
• def → run safely in background thread using await asyncio.to_thread(...)

# List all registered jobs fastapi-crons list # Manually run a specific job fastapi-crons run_job <job_name>

• Distributed locking via Redis
• Retry policies
• Manual run triggers via HTTP
• Admin dashboard with metrics

You can add tags to jobs for better organization:

@cron_job("*/5 * * * *", tags=["maintenance", "cleanup"]) async def cleanup_job(): # This job has tags for categorization pass

FastAPI App │ ├── Crons() │ ├── Registers decorated jobs │ ├── Starts background scheduler (async) │ ├── SQLite Backend │ ├── Tracks last run for each job │ ├── /crons endpoint │ ├── Shows current job status (with timestamps) │ └── CLI Tool ├── List jobs / Run manually 

We welcome PRs and suggestions! If you'd like this added to FastAPI officially, fork the repo, polish it, and submit to FastAPI with a clear integration proposal.

• Each job has an isolated error handler
• Errors are printed and don't block scheduler
• Future: Add error logging / alert hooks

Licence

Email me

github

Made with ❤️ by Mehar Umar.
Designed to give developers freedom, flexibility, and control when building production-grade FastAPI apps.