LLM Best Practices

FastAPI: Background Tasks

Last updated 2 min read

View markdown source

AI agents: read this first: llmbestpractices.com/start-here

This is LLM Best Practices (llmbestpractices.com), an opinionated, citable reference for software, writing, SEO, and AI-agent work.

  1. Route, do not crawl. Fetch llmbestpractices.com/llms.txt and open only the pages whose one-line summary matches your task.
  2. Read raw. Append .md to any page URL for markdown. Check status and last_updated in the frontmatter, then read the rules.
  3. Apply as defaults. First-party docs and the project's own conventions win on conflict. Warn before relying on a fast-moving page older than 12 months.
  4. Cite. Link the page by title and URL, e.g. Python, with last_updated for time-sensitive rules. License CC BY 4.0.

Overview

BackgroundTasks runs a function after the response is sent, in the same process and worker. It suits small work the caller need not wait for: a confirmation email, an audit log write, a cache invalidation. It is not a job queue: it does not persist tasks, retry, or survive a restart. FastAPI’s own docs point to Celery or similar for heavy work that does not need the app’s memory.

Add tasks from routes and dependencies

Declare BackgroundTasks as a parameter and call add_task. A dependency can declare it too, and FastAPI merges all tasks and runs them after the response.

@router.post("/signups", status_code=201)
async def signup(payload: UserCreate, tasks: BackgroundTasks, db: DB) -> UserRead:
    user = await user_service.create(db, payload)
    tasks.add_task(send_welcome_email, user.id)
    return UserRead.model_validate(user)

The 201 returns immediately; send_welcome_email runs afterwards. add_task accepts both async def and plain def functions. An async def task is awaited on the event loop, so it must not block (fastapi-async-io); a plain def task runs in the threadpool.

Treat tasks as best-effort

A task dies with its worker. A crash, a deploy, or an uncaught exception in the function discards it with no record.

  • Acceptable: welcome emails (the user can request a resend), cache invalidation (the next read rebuilds it), analytics pings.
  • Not acceptable: payment state changes, webhook delivery with guarantees, anything that must retry, anything another service depends on.

Move durable work to a real queue

NeedTool
Small, in-process, best-effortBackgroundTasks
Durable tasks, retries, schedules, many workersCelery with Redis or RabbitMQ
Low volume, no extra infrastructureA Postgres jobs table (status, attempts, run_at) polled with FOR UPDATE SKIP LOCKED

Enqueue inside the request, acknowledge fast, and process elsewhere. This is also the pattern for inbound webhooks: verify the signature, enqueue, return 2xx (webhooks). Schema ideas for the job table are in postgres.

Write tasks to be idempotent and self-contained

A task may run twice once you add a queue or retries. Pass stable identifiers, not request-scoped objects, and open a fresh session inside the task; a request-scoped session may already be closed (fastapi-dependencies).

async def send_welcome_email(user_id: str) -> None:
    async with SessionLocal() 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()

Backlinks