Overview
FastAPI’s BackgroundTasks runs a function after the HTTP response is sent. It is the right tool for lightweight fire-and-forget work: sending a confirmation email, writing an audit log entry, or invalidating a cache key. It is not a job queue. It does not persist tasks, retry on failure, or survive a worker restart. Understanding both sides of that trade-off prevents using BackgroundTasks where a real queue is required. See fastapi for the brief overview; this page covers the mechanics, limits, and upgrade path.
Use BackgroundTasks for work the caller need not await
Inject BackgroundTasks directly into a route or dependency and call add_task before returning the response.
from fastapi import APIRouter, BackgroundTasks, Depends
router = APIRouter()
@router.post("/signups", response_model=UserRead, status_code=201)
async def signup(
payload: UserCreate,
tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db),
) -> UserRead:
user = await user_service.create(db, payload)
tasks.add_task(send_welcome_email, user.email, user.name)
return UserRead.model_validate(user)The response returns with 201 immediately. send_welcome_email runs in the same worker process after the response is flushed. The caller never waits on it.
Keep background functions async when they do I/O
add_task accepts both async def and plain def functions. For any function that makes network calls or database writes, use async def. FastAPI will await it on the event loop rather than blocking in a thread. See fastapi-async-io for why mixing blocking code into async context causes problems.
async def send_welcome_email(to: str, name: str) -> None:
async with httpx.AsyncClient() as client:
await client.post(
"https://mail.example.com/send",
json={"to": to, "subject": f"Welcome, {name}"},
)For CPU-bound background work, use asyncio.to_thread inside the async def task rather than a plain def, so the call signature stays consistent.
Inject BackgroundTasks into dependencies, not just routes
Dependencies can accept BackgroundTasks the same way routes do. This is useful for cross-cutting concerns like audit logging.
from fastapi import BackgroundTasks, Depends
async def audit_log(
action: str,
user: User = Depends(get_current_user),
tasks: BackgroundTasks = Depends(),
) -> None:
tasks.add_task(write_audit_entry, user.id, action)
@router.delete("/{item_id}", dependencies=[Depends(partial(audit_log, "delete_item"))])
async def delete_item(item_id: str, db: AsyncSession = Depends(get_db)) -> None: ...FastAPI merges background tasks from all dependencies and the route body into a single list and runs them all after the response. See fastapi-dependencies for the dependency injection model.
Treat BackgroundTasks as best-effort, not guaranteed delivery
Tasks added via BackgroundTasks run in the uvicorn worker process. If the worker crashes, is killed by a deployment, or hits an uncaught exception in the task function, the task is gone with no record. This is acceptable for:
- Welcome and transactional notification emails (the user can request a resend).
- Cache invalidation (the next request rebuilds it).
- Soft analytics pings (data loss is tolerable).
It is not acceptable for:
- Payment processing or financial state changes.
- Webhook delivery with SLA guarantees.
- Work that must retry on network failure.
- Work that another service depends on completing.
For durable work, use a real queue.
Reach for Celery or Arq when durability matters
| Need | Tool |
|---|---|
| Simple fire-and-forget, in-process | BackgroundTasks |
| Durable tasks, retries, priorities | Celery + Redis/RabbitMQ |
| Async-native, lightweight, Redis-backed | Arq |
| Database-backed, no extra infra | Postgres job table + polling |
Celery is the most widely deployed Python task queue. Arq is a good fit when the stack is already async and Redis is available. A database job table (jobs with status, attempts, run_at) is a valid choice for low-volume work where adding a broker is not worth the operational overhead. See postgres for the schema pattern.
Design background task functions for idempotency
Background tasks may be called more than once if you migrate to a real queue later or if you add retry logic. Write task functions so that running them twice produces the same result as running them once.
async def send_welcome_email(user_id: str) -> None:
async with get_session() as db:
user = await db.get(User, user_id)
if user.welcome_sent:
return # idempotency guard
await mailer.send_welcome(user.email)
user.welcome_sent = True
await db.commit()Pass a stable identifier (like user_id) rather than a mutable object to the task. This avoids capturing stale state from the request and makes the guard check reliable. See python-async for the async session pattern used in a background context outside a request.