Sentry Answers>FastAPI>

FastAPI Background Tasks and Middleware

FastAPI Background Tasks and Middleware

David Y.

The ProblemJump To Solution

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:

Click to Copy
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.

Click to Copy
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.

Click to Copy
# 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.
Click to Copy

Using Background Tasks in FastAPI

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

Click to Copy
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:
Click to Copy

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:

Click to Copy
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:

Click to Copy
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.

  • Community SeriesIdentify, Trace, and Fix Endpoint Regression Issues
  • ResourcesBackend Error Monitoring 101
  • Syntax.fm logo
    Listen to the Syntax Podcast

    Tasty treats for web developers brought to you by Sentry. Get tips and tricks from Wes Bos and Scott Tolinski.

    SEE EPISODES

Loved by over 4 million developers and more than 90,000 organizations worldwide, Sentry provides code-level observability to many of the world’s best-known companies like Disney, Peloton, Cloudflare, Eventbrite, Slack, Supercell, and Rockstar Games. Each month we process billions of exceptions from the most popular products on the internet.

© 2024 • Sentry is a registered Trademark
of Functional Software, Inc.