# FastAPI: Schedule Recurring Tasks with APScheduler

> Schedule recurring tasks in FastAPI using APScheduler with CronTrigger or IntervalTrigger, and shut the scheduler down cleanly with FastAPI's lifespan

**URL:** https://sentry.io/answers/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](https://pypi.org/project/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:

```sh
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:

```python
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`](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html#module-apscheduler.triggers.cron), which uses a syntax similar to the nix [`cron` scheduler](https://en.wikipedia.org/wiki/Cron) to specify when a task should be executed. Finally, we use FastAPI's [Lifespan](https://fastapi.tiangolo.com/advanced/events/) 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:

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

See [the examples in the APScheduler documentation](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html#examples) to learn more about how to set different schedules.

Alternatively, use an [`IntervalTrigger`](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/interval.html#module-apscheduler.triggers.interval) to achieve the same thing:

```python
# 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)
```

---

*Source: [sentry.io/answers/schedule-tasks-with-fastapi/](https://sentry.io/answers/schedule-tasks-with-fastapi/)*