How to Schedule Background Functions

To schedule functions for execution after a response has been sent to the client, use the BackgroundTasks class from fastapi. This allows you to perform operations like sending emails, processing files, or updating logs without making the user wait for the task to complete.

Schedule a Task in a Path Operation

To schedule a background task, declare a parameter of type BackgroundTasks in your path operation function and call its add_task method.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message: str = ""):
    with open("log.txt", mode="a") as email_file:
        content = f"notification for {email}: {message}\n"
        email_file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # Schedule the function to run after the response is sent
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

In this example, write_notification is a standard def function. FastAPI will run it in a separate thread to avoid blocking the event loop.

Schedule Tasks within Dependencies

You can also inject BackgroundTasks into dependency functions. FastAPI ensures that the same BackgroundTasks instance is shared across the entire request, allowing both dependencies and the path operation to contribute to the same queue.

from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q

@app.post("/send-notification/{email}")
async def send_notification(
    email: str, 
    background_tasks: BackgroundTasks, 
    q: Annotated[str | None, Depends(get_query)] = None
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

Using Async Functions

The add_task method supports both regular def functions and async def functions. If you provide an async def function, FastAPI will await it directly in the event loop after the response is sent.

import asyncio
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

async def async_cleanup(filename: str):
    await asyncio.sleep(5)  # Simulate long-running async work
    print(f"Cleaned up {filename}")

@app.post("/process/{filename}")
async def process_file(filename: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(async_cleanup, filename)
    return {"message": "Processing started"}

Key Implementation Details

• Execution Timing: Functions added via add_task are executed after the response has been sent to the client. This means you cannot modify the response (headers, status code, or body) from within a background task.
• Class Selection: Always use fastapi.background.BackgroundTasks (plural). While Starlette provides a singular BackgroundTask, the FastAPI plural version is designed to work seamlessly with the dependency injection system, allowing multiple tasks to be queued during a single request.
• Parameter Passing: The add_task method accepts the function as the first argument, followed by any positional and keyword arguments that should be passed to that function: background_tasks.add_task(func, *args, **kwargs).
• Dependency Sharing: As shown in tests/test_response_dependency.py, using Annotated[BackgroundTasks, Depends(...)] allows you to wrap background task logic into reusable dependencies while still having access to the task queue in your main endpoint.