Running Scheduled Jobs in FastAPI

Aug 7, 2024 • 3 min read

Question: I want to schedule a task to run once a day at 01:05 pm UTC in my FastAPI application. How can I achieve this?

One popular library for running scheduled jobs in Python is APScheduler. It’s a light but powerful in-process task scheduler that lets you schedule functions to be executed at times of your choosing.

Lets say you have the following FastAPI code:

Python 
# main.py

from fastapi import FastAPI


app = FastAPI()


@app.get('/')
async def home():
    return {'message': 'API is up and running'}


@app.get('/say_hello')
async def say_hello(name: str = 'World'):
    return {'message': f'Hello, {name}!'}

And you want it to run an async task at specific time once a day. For example, you want to make asynchronous HTTP request to a specific API:

Python 
import aiohttp 

async def fetch_current_time():
    url = 'https://timeapi.io/api/Time/current/zone?timeZone=UTC'
    async with aiohttp.ClientSession() as session:
        try:
            r = await session.get(url)
            r.raise_for_status()
            print(f'TimeAPI result: {await r.json()}')
        except aiohttp.ClientError as e:
            print(f'Error: {e}')

To use APScheduler, first you need to install it using pip:

pip install apscheduler

Configure the scheduler:

Python 
# main.py 

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from pytz import utc

... 

scheduler = AsyncIOScheduler(timezone=utc)

Start the scheduler in the FastAPI’s Lifespan Events:

Python 
# main.py 

from contextlib import asynccontextmanager

... 

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    yield
    scheduler.shutdown()


app = FastAPI(lifespan=lifespan)

...

Then add the scheduled_job decorator to your function:

Python 
@scheduler.scheduled_job('cron', hour=13, minute=05)
async def fetch_current_time():
    ...

That’s it! Run your application and you will see it will execute the scheduled task at 01:05 pm UTC once a day:

uvicorn main:app --reload

Putting it all together

The complete script is shown below:

Python 
# main.py 

from contextlib import asynccontextmanager

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from pytz import utc
import aiohttp


scheduler = AsyncIOScheduler(timezone=utc)


@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    yield
    scheduler.shutdown()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def home():
    return {'message': 'API is up and running'}


@app.get('/say_hello')
async def say_hello(name: str = 'World'):
    return {'message': f'Hello, {name}!'}


@scheduler.scheduled_job('cron', hour=13, minute=05)
async def fetch_current_time():
    url = 'https://timeapi.io/api/Time/current/zone?timeZone=UTC'
    async with aiohttp.ClientSession() as session:
        try:
            r = await session.get(url)
            r.raise_for_status()
            print(f'TimeAPI result: {await r.json()}')
        except aiohttp.ClientError as e:
            print(f'Error: {e}')

What if I want to run a task at a fixed interval?

You can also configure the scheduler to run your jobs at a fixed interval or maybe just once at a certain point of time. Some examples:

Python 
# Run once on Aug 7, 2024 at 10:30 AM
@scheduler.scheduled_job('date', run_date='2024-08-07 10:30:00')

# Run every 10 seconds
@scheduler.scheduled_job('interval', seconds=10)

# Run every weekday (monday to friday) at 7:45 AM until Dec 31, 2024
@scheduler.schedule_job(
    'cron', 
    day_of_week='mon-fri', 
    hour=7, 
    minute=45, 
    end_date='2024-12-31'
)

You can find more information about configuring the scheduler in the docs.