Schedule tasks with FastAPI

The Problem

In my FastAPI project, I have a task that needs to run once a day, at midnight. How should I implement this?

The Solution

Use the Advanced Python Scheduler library, APScheduler. You can schedule tasks for execution at specified times, such as every day at midnight, at regular intervals like every four hours, or once off at exactly 3:14 AM on January 19, 2038. Call the scheduler from the FastAPI code and the task will run on schedule as long as the program is running.

First, install the library. If you’re using a virtual environment, make sure it’s activated in your shell before running this command:

pip install apscheduler

When APScheduler is installed, integrate it with your existing code. Here is an example of a minimal FastAPI application with a scheduled task:

from fastapi import FastAPI
from datetime import datetime
from contextlib import asynccontextmanager
from apscheduler.schedulers.background import BackgroundScheduler  # runs tasks in the background
from apscheduler.triggers.cron import CronTrigger  # allows us to specify a recurring time for execution

# The task to run
def my_daily_task():
    print(f"Task is running at {datetime.now()}")
    # ... additional task code goes here ...

# Set up the scheduler
scheduler = BackgroundScheduler()
trigger = CronTrigger(hour=0, minute=0)  # midnight every day
scheduler.add_job(my_daily_task, trigger)
scheduler.start()

app = FastAPI()

# Ensure the scheduler shuts down properly on application exit.
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    scheduler.shutdown()

@app.get("/")
def read_root():
    return {"message": "FastAPI with APScheduler Demo"}

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

In this example, we create and start a BackgroundScheduler object. This object can handle multiple tasks at different times, specified by triggers. As we want our task to run every day at midnight, we use a CronTrigger, which uses a syntax similar to the nix cron scheduler to specify when a task should be executed. Finally, we use FastAPI’s Lifespan manager to ensure that the scheduler shuts down when the application is stopped.

The CronTrigger class is flexible enough that we can schedule jobs to run daily, on the hour, or even at specific times, like 12.13 pm every second Tuesday, as in this example:

CronTrigger(day='2nd tue', hour=12, minute=13)

See the examples in the APScheduler documentation to learn more about how to set different schedules.

Alternatively, use an IntervalTrigger to achieve the same thing:

# Set up the scheduler
from contextlib import asynccontextmanager

from fastapi import FastAPI
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler  # runs tasks in the background
from apscheduler.triggers.interval import IntervalTrigger  # <-- CHANGED IMPORT

# The task to run
def my_daily_task():
    print(f"Task is running at {datetime.now()}")
    # ... additional task code goes here ...

# Set up the scheduler
scheduler = BackgroundScheduler()
trigger = IntervalTrigger(hours=24,
                          start_date='2025-01-01 00:00:00')  # <-- CHANGED LINE (run every 24 hours, starting in 2025)
scheduler.add_job(my_daily_task, trigger)
scheduler.start()

app = FastAPI()

# Ensure the scheduler shuts down properly on application exit.
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    scheduler.shutdown()

@app.get("/")
def read_root():
    return {"message": "FastAPI with APScheduler Demo"}

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)