Use background tasks in FastAPI

Run lightweight work after sending a response, pass safe task inputs, and recognize when work belongs in a durable queue.

Guance Editorial Team 2026-10-27 3 min read Updated 2026-10-27

Best practices

FastAPI's BackgroundTasks runs work after the response is sent, inside the same web process. It is useful for small follow-up actions, but it does not guarantee eventual completion. Use a durable task system when losing work on a process restart is unacceptable.

Reduce response waiting

A request can return “accepted” before a follow-up action finishes. This reduces the client's wait, but the work still consumes server resources and does not automatically improve throughput.

Register a task

from fastapi import FastAPI, BackgroundTasks, UploadFile

app = FastAPI()

def send_welcome_email(email: str):
    # Placeholder for the application's mail integration.
    print(f"Sending a welcome email to {email}")

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_welcome_email, email)
    return {"msg": "Signup accepted; welcome email requested"}

This example demonstrates task registration, not a complete signup or mail-delivery implementation. FastAPI supplies the BackgroundTasks object. Tasks registered on one response run in registration order after the response has been sent.

Pass application data

The callable can accept ordinary Python values:

def generate_report(user_id: int, period: str, recipients: list[str]):
    report = build_report(user_id, period)
    for recipient in recipients:
        deliver(report, recipient)

@app.post("/reports")
def create_report(req: ReportRequest, background_tasks: BackgroundTasks):
    background_tasks.add_task(
        generate_report, req.user_id, req.period, req.recipients
    )
    return {"msg": "Report requested", "status": "processing"}

ReportRequest, build_report(), and deliver() are application-defined placeholders. Passing IDs rather than large objects often keeps dependencies clearer: the task can acquire its own database session and fetch the needed data. BackgroundTasks itself does not serialize these arguments, and request-scoped resources may already be closed when the task runs.

Handle an uploaded file

@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
    path = await save_temp(file)
    background_tasks.add_task(scan_and_process, path)
    return {"filename": file.filename, "status": "accepted"}

Supply save_temp() and scan_and_process() in the application. The endpoint still waits for the upload to be saved before returning. Pass a durable path or object-storage key to the task instead of relying on an UploadFile object that may be closed after the request.

Set size limits, clean up temporary data, and record failed or incomplete processing. Progress polling or WebSocket notifications require a separate application-level implementation.

Decide when to use a queue

Requirement	Typical approach
Small best-effort follow-up action	`BackgroundTasks`
Durable work with retries and failure handling	A properly configured queue such as Celery, ARQ, or Dramatiq
Execution across processes or machines	A task queue and workers
Scheduled execution	APScheduler or a queue's scheduling component
Long-running or heavy CPU work	Separate workers or processes

Unfinished in-process tasks can be lost on restart. A persistent queue still needs acknowledgments, retry policy, failure handling, and idempotent effects; it does not automatically provide exactly-once execution.

Observe task outcomes

A background failure cannot change an HTTP response already sent. Log task start, completion, failure, and elapsed time with a shared task ID. Store business state for pending work so that missing completion can be distinguished from success.

Guance log detection can alert on collected failure records. Detecting work that never completed also requires the expected-task state or a suitably defined completion window.

FAQ

Should the task use def or async def?

A synchronous def task runs in a thread pool; an async def task runs on the event loop. Use asynchronous I/O inside the latter. Blocking calls inside an async function still block the loop. Heavy CPU work belongs in separate processes or workers.

What happens if a task raises?

The exception propagates on the server side after the response has been sent, and later tasks in that same BackgroundTasks sequence are not executed. Record failures explicitly and handle expected exceptions where the business workflow requires later steps to continue. Do not silently convert failure into success.

Can clients query progress?

There is no built-in progress store. Write progress and outcomes to a database or Redis and expose a status endpoint; use WebSockets if real-time delivery is required.

References: FastAPI background tasks, Starlette background tasks.