TopolearnUsers shouldn't wait for a notification email or an audit log write before they get their response.
Declare a parameter of type BackgroundTasks and FastAPI injects an instance. Call background_tasks.add_task(func, *args, **kwargs) and return as usual: the response goes out first, then the tasks run in the order they were added, in the same process. func can be a plain def (run in the thread pool) or an async def (run on the event loop).
BackgroundTasks also works in dependencies. A dependency can declare it and add tasks, and FastAPI merges everything into one list for the request, which suits things like "write an access log entry" inside an auth dependency.
Know the limits, because this is a convenience, not a job queue:
- Tasks live in memory. If the process restarts or crashes, pending tasks are lost, and nothing retries them.
- They run in the worker that served the request and share its resources. A heavy task slows that worker, and a blocking call inside an
asynctask freezes its event loop. - The client never learns whether a task succeeded; exceptions are only logged.
So use them for short, best-effort side effects: a notification, an audit record, a cache invalidation. For work that must happen, takes minutes, needs retries or should run on other machines, use a real queue (Celery, RQ, Dramatiq, Arq or a cloud queue): the endpoint enqueues a job and returns its id, often with status 202 Accepted.
Pass tasks plain values such as an email address or a row id, not request-scoped objects like the database session from a dependency. Open whatever the task needs, such as a fresh session, inside the task itself.
Example
from fastapi import BackgroundTasks, Depends, FastAPI
from fastapi.testclient import TestClient
log: list[str] = []
def write_log(message: str) -> None:
log.append(message) # pretend this is slow I/O
def audit(background_tasks: BackgroundTasks, q: str | None = None):
if q:
background_tasks.add_task(write_log, f"searched {q!r}")
app = FastAPI()
@app.post("/notify/{email}", status_code=202, dependencies=[Depends(audit)])
def notify(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"mail sent to {email}")
return {"queued": True}
r = TestClient(app).post("/notify/ann@example.com?q=shoes")
print(r.json(), log)
# {'queued': True} ["searched 'shoes'", 'mail sent to ann@example.com']Remember
- Add a BackgroundTasks parameter and call add_task(func, *args)
- Tasks run after the response is sent, in the same process
- Dependencies can add tasks too; they're merged per request
- Use a real queue for anything that must not be lost or needs retries
Watch out for
- Passing the request's database session to a task instead of opening a new one inside it.