FastAPI: Create a repeated task

sometimes, you want a task to trigger not just when the server starts but also periodically. For example, you might want to regularly check the availability of some endpoints, send a heartbeat, to do a data refresh from a database or another API.

You can accomplish this by triggering a loop inside a start-up event, but there are a few challenges to overcome:

1. You finish the startup event before the periodic loop ends (so the server can start!)
2. If the repeated tasks perform blocking IO, it shouldn’t block the event loop.
3. Exceptions raised by the periodic task shouldn’t just be silently swallowed.

The fastapi_utils.tasks.repeat_every decorator handles all of these issues and adds some other conveniences.

To install fastapi_utils enter in a console

pip3 install fastapi-utils

The @repeat_every decorator

When a function decorated with the @repeat_every(...) decorator is called, a loop is started, and the function is called periodically with a delay determined by the seconds argument provided to the decorator.

If you also apply the @app.event("startup") decorator, FastAPI will call the function during server startup, and the function will then be called repeatedly while the server is still running.

Here’s a hypothetical example that could be used to print to console “hello world” every 30 seconds.

#!/usr/bin/env python3 
from fastapi import FastAPI, APIRouter 
from fastapi_utils.tasks import repeat_every 
import uvicorn 
 
app = FastAPI( title="Recurring task example.", openapi_url="/openapi.json") 
api_router = APIRouter() 
 
@app.on_event("startup") 
@repeat_every(seconds=30) 
def test_print(): 
    """ 
    Test print 
    """ 
    print("Hello world") 
 
app.include_router(api_router) 
 
if __name__ == "__main__": 
    uvicorn.run(app, host="0.0.0.0", port=8001, log_level="debug")