FastAPI Background Tasks and Middleware

David Y.

—January 15, 2024

The Problem

How do I use a FastAPI background task within custom FastAPI middleware?

The Solution

Normally to trigger background tasks in FastAPI, you can simply pass in a BackgroundTasks object to your function. However, when you are using middleware it’s harder to customize what kind of functions you can pass in. Instead you can create the BackgroundTasks object manually and assign it to your response by using response.background = background_tasks. A complete example is as follows:

from fastapi import FastAPI
from fastapi import BackgroundTasks
from fastapi import Request

app = FastAPI()

def my_background_task():
    print("runs in background")

@app.middleware("http")
async def my_background_task_middleware(request: Request, call_next):
    background_tasks = BackgroundTasks()
    background_tasks.add_task(my_background_task)
    response = await call_next(request)
    response.background = background_tasks
    return response

@app.get("/")
def hello():
    return {"message": "Hello World"}

Below, we describe how background tasks and middleware work in FastAPI in more detail, both separately and in combination.

FastAPI Background Tasks

Sometimes you want to execute some code without delaying the response to the end user. For example, after a user signs up to your web application, you might want to send a “Welcome” email and show a confirmation page to the user. There’s no need to make the user wait until the email has been sent before showing the confirmation page — that can happen in the background.

We can start with a basic “Hello World” application and add a long-running background task. We’ll simulate the task with a sleep(5), which will simply wait for five seconds and then print out a status message, but in a real application this code would be doing something useful instead.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def hello():
    return {"message": "Hello World"}

A naive way to add the task is simply to call the function before returning a response to the user.

Don’t do this, as it will make the user wait for five seconds unnecessarily.

# Don't do this - the user will have to wait
from fastapi import FastAPI
from time import sleep

app = FastAPI()

def my_longrunning_background_function(status: str):
    sleep(5)
    print(f"\n\n\n---\nAll done, status {status}\n---\n\n\n")

@app.get("/")
async def hello():
    my_longrunning_background_function("my status")
    return {"message": "Hello World"}

If you load the page, you’ll see the following things happen:

• The page hangs for five seconds.
• Then the response {"message":"Hello World"} is shown in the web browser.
• The following output is shown in the console.

Using Background Tasks in FastAPI

Instead of the above, you can execute the long-running task in the background without making the user wait:

from fastapi import FastAPI
from fastapi import BackgroundTasks
from time import sleep

app = FastAPI()

def my_longrunning_background_function(status: str):
    sleep(5)
    print(f"\n\n\n---\nAll done, status {status}\n---\n\n\n")

@app.get("/")
async def hello(background_tasks: BackgroundTasks):
    background_tasks.add_task(my_longrunning_background_function, status="my status")
    return {"message": "Hello World"}

Here we pass a BackgroundTasks object to our hello() route. Instead of executing our task directly, it adds it to the background tasks where it will be automatically picked up by FastAPI as soon as possible. Without waiting we return the response to the user.

This will result in the following behavior when you load the page:

• The user immediately sees the {"message": "Hello World"} in the browser.
• After five seconds, you see the following output in the console:

Using Middleware in FastAPI

Middleware is code that automatically gets applied to your main routes without you having to explicitly add it each time. It’s often used for modifying HTTP requests, logging, or any other case where you need to apply the same logic to all (or some category of) existing functions, and functions that you might add in the future.

Here’s an example of a basic custom middleware in FastAPI where we print "I executed" each time a route is visited:

from fastapi import FastAPI
from fastapi import Request

app = FastAPI()

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    response = await call_next(request)
    print("I executed")
    return response

@app.get("/")
async def hello():
    return {"message": "Hello World"}

@app.get("/goodbye")
async def goodbye():
    return {"message": "Goodbye World"}

Note that now we’ve added two routes to our app — hello() and goodbye(). We don’t call the middleware function explicitly from either, but if you visit either route you’ll see "I executed" printed to the logs because the app.middleware("http") decorator says that we want this middleware to execute for all HTTP requests.

The middleware has access to the Request object. We generate the response that the calling function would have generated by using call_next(request) and we can modify or use information from either the request or the response, or execute any other arbitrary code, before returning the final response to the user.

Combining Middleware and Background Tasks

Now that you’ve seen both middleware and background tasks in action, the next example where we combine them will be clearer. Here’s a longer example where we again use sleep to simulate a background task and have it execute for every route in our application:

from fastapi import FastAPI
from fastapi import BackgroundTasks
from fastapi import Request
from time import sleep

app = FastAPI()

@app.middleware("http")
async def my_longrunning_background_function_middleware(request: Request, call_next):
    background_tasks = BackgroundTasks()
    background_tasks.add_task(my_longrunning_background_function, status="my status")
    response = await call_next(request)
    response.background = background_tasks
    return response

def my_longrunning_background_function(status: str):
    sleep(5)
    print(f"\n\n\n---\nAll done, status {status}\n---\n\n\n")

@app.get("/")
async def hello():
    return {"message": "Hello World"}

@app.get("/goodbye")
async def goodbye():
    return {"message": "Goodbye World"}

As in our first example, we manually create a BackgroundTasks object within our middleware. We then add our long-running background function to the list of tasks, attach this to the response object, and then return the response to the user.

For any route the user visits, they will see the response in the browser immediately, but the status message will only appear in the console output five seconds later.