Question 1 / 12
How do BackgroundTasks work in FastAPI?
BackgroundTasks in FastAPI allow you to run functions after the response has been sent to the client. They are built on top of Starlette's BackgroundTask class.
How it works:
- Declare
BackgroundTasksas a parameter in your endpoint (FastAPI injects it automatically) - Call
background_tasks.add_task(func, *args, **kwargs)to schedule work - The response is sent immediately, and the tasks run afterward in the same process
Key characteristics:
- Tasks run in the same process as the request (no separate worker)
- They execute after the response is sent
- Both sync and async functions are supported
- Tasks run sequentially in the order they were added
- If the task is a sync function, it runs in a thread pool; async functions run on the event loop
When to use:
- Sending emails or notifications
- Writing audit logs
- Light post-processing (thumbnails, cache warming)
- Any work that doesn't need to be in the response
Pseudo Code
from fastapi import BackgroundTasks
def do_work(data):
# runs after response is sent
process(data)
@app.post("/items/")
async def create_item(background_tasks: BackgroundTasks):
background_tasks.add_task(do_work, data)
return {"status": "accepted"}Code
from fastapi import FastAPI, BackgroundTasks
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
def send_email(to: str, subject: str, body: str):
"""Simulates sending an email (runs after response)."""
logger.info(f"Sending email to {to}: {subject}")
time.sleep(2) # Simulate slow email service
logger.info(f"Email sent to {to}")
def write_audit_log(action: str, user_id: int):
"""Simulates writing an audit log."""
logger.info(f"Audit: user {user_id} performed '{action}'")
@app.post("/register")
async def register_user(
background_tasks: BackgroundTasks,
):
# User is created immediately
user = {"id": 42, "email": "alice@example.com"}
# These run AFTER the response is sent
background_tasks.add_task(
send_email,
to=user["email"],
subject="Welcome!",
body="Thanks for registering.",
)
background_tasks.add_task(
write_audit_log,
action="register",
user_id=user["id"],
)
# Client gets this immediately, without waiting for email
return {"id": user["id"], "status": "registered"}
# Async background task
async def update_search_index(item_id: int):
"""Async tasks run on the event loop."""
import asyncio
await asyncio.sleep(1) # Simulate async work
logger.info(f"Search index updated for item {item_id}")
@app.post("/items/")
async def create_item(background_tasks: BackgroundTasks):
item_id = 99
background_tasks.add_task(update_search_index, item_id)
return {"id": item_id, "status": "created"}Key Points
- •BackgroundTasks run after the response is sent, in the same process
- •Both sync and async functions are supported -- sync runs in a threadpool
- •Tasks execute sequentially in the order they were added
- •FastAPI automatically injects BackgroundTasks when declared as a parameter
Gotchas
- •Background tasks share the same process -- a CPU-heavy task will block the event loop for async tasks
- •If the server crashes, queued background tasks are lost (no persistence)
- •Exceptions in background tasks are not reported to the client since the response is already sent
- •Do not use BackgroundTasks for long-running jobs -- use Celery or ARQ instead