Intermediate
Your user hits “Register” and your API needs to create their account, send a welcome email, log the event, and maybe kick off an onboarding workflow. If you do all of that before returning a response, the user is staring at a spinner for two seconds while your email server thinks about things. That is a bad experience — and it is completely unnecessary. The account creation is what matters. Everything else can happen after you say “201 Created.”
FastAPI ships with a built-in BackgroundTasks class that does exactly this. You declare a task, hand it to FastAPI, and it runs after the response is sent to the client. No Redis. No worker process. No Celery config file. Just a Python function that runs in the background while the user already has their answer. It is the right tool when your background work is lightweight and you do not need a separate process or message queue.
This article covers everything you need to use BackgroundTasks effectively: the basic pattern, passing arguments, running multiple tasks, using it with dependency injection, handling async tasks, and a real-world example that ties it all together. We also cover when you should reach for Celery instead — because BackgroundTasks is not the answer to every problem.
FastAPI BackgroundTasks: Quick Example
Here is the minimal working version. We create a FastAPI route, inject the BackgroundTasks object, add a task to it, and return immediately. FastAPI sends the response first, then runs the task.
# quick_background.py
from fastapi import FastAPI, BackgroundTasks
import time
app = FastAPI()
def write_log(message: str):
# Simulates slow I/O -- runs AFTER the response is sent
time.sleep(1)
with open("app.log", "a") as f:
f.write(message + "\n")
@app.post("/register")
def register_user(name: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"New user registered: {name}")
return {"message": f"Welcome, {name}! Account created."}
Output (client sees immediately, no delay):
{"message": "Welcome, Alice! Account created."}
The key line is background_tasks.add_task(write_log, f"New user registered: {name}"). The first argument is the function to call; everything after is passed as arguments to that function. FastAPI handles the scheduling — the function runs in the same process, in the same event loop, but only after the HTTP response has been dispatched.
The sections below cover how to use this pattern with more complex real-world requirements: multiple tasks, dependency injection, async functions, and error handling strategies.
What is BackgroundTasks and When Should You Use It?
BackgroundTasks is a lightweight task deferral mechanism built into the Starlette framework that FastAPI is built on. When you call add_task(), Starlette queues the function internally and runs it after the response generator has finished sending data to the client. The task runs in the same worker process — there is no separate thread pool or event loop by default.
Think of it like a restaurant waiter who takes your order, delivers it to the kitchen, then hands you your receipt — and then goes to refill the water glasses. The customer has their receipt immediately. The background work (refilling water) happens afterward without making the customer wait.
This also means BackgroundTasks has real limitations. Use the comparison table below to pick the right tool for your situation:
| Scenario | BackgroundTasks | Celery / ARQ |
|---|---|---|
| Send a welcome email | Good fit | Overkill |
| Write a log entry | Good fit | Overkill |
| Process an uploaded image (3 sec) | Acceptable | Better choice |
| Generate a video (60+ sec) | Bad fit — ties up worker | Required |
| Retry on failure | Not supported | Built-in |
| Scheduled/periodic tasks | Not supported | Celery Beat |
| Multiple workers / scale out | Not supported | Required |
| Zero infrastructure overhead | Yes | No (Redis/RabbitMQ) |
If your background work is fast (under a few seconds), runs in the same process, and does not need retries or scheduling, BackgroundTasks is the cleanest solution. For anything heavier, reach for a real task queue.
Adding Tasks With Arguments
The add_task() method accepts positional and keyword arguments after the function reference. Any serializable value works — strings, integers, dictionaries, lists. You pass them directly; FastAPI stores them and calls the function with them after the response is sent.
# background_args.py
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def send_email(to: str, subject: str, body: str):
# In production this calls your SMTP library or email service
print(f"[EMAIL] To: {to} | Subject: {subject}")
print(f"[EMAIL] Body: {body[:50]}...")
def log_event(event_type: str, user_id: int, metadata: dict):
print(f"[LOG] {event_type} | user={user_id} | meta={metadata}")
@app.post("/purchase/{user_id}")
def complete_purchase(user_id: int, item: str, background_tasks: BackgroundTasks):
# Queue two independent background tasks
background_tasks.add_task(
send_email,
to=f"user{user_id}@example.com",
subject="Order Confirmed",
body=f"Your order for {item} has been placed successfully."
)
background_tasks.add_task(
log_event,
event_type="purchase",
user_id=user_id,
metadata={"item": item, "source": "web"}
)
return {"status": "ok", "item": item}
Output (client sees instantly):
{"status": "ok", "item": "Python Handbook"}
Background output (printed after response):
[EMAIL] To: user42@example.com | Subject: Order Confirmed
[EMAIL] Body: Your order for Python Handbook has been placed succe...
[LOG] purchase | user=42 | meta={'item': 'Python Handbook', 'source': 'web'}
Tasks run in the order they are added. If the first task raises an unhandled exception, subsequent tasks in the same request will not run — we will cover error handling strategies in a later section.
Background Tasks with Dependency Injection
FastAPI’s dependency injection system and BackgroundTasks work naturally together. You can inject BackgroundTasks into a dependency function the same way you inject it into a route. This is useful when you want a service class or utility function to schedule its own cleanup or notification work without the route handler knowing about the details.
# background_dependency.py
from fastapi import FastAPI, BackgroundTasks, Depends
from dataclasses import dataclass
app = FastAPI()
# Simulated database
user_db: dict = {}
@dataclass
class UserService:
background_tasks: BackgroundTasks
def create_user(self, name: str, email: str) -> dict:
user_id = len(user_db) + 1
user_db[user_id] = {"name": name, "email": email}
# Schedule side effects -- the route handler doesn't need to know
self.background_tasks.add_task(self._send_welcome, email, name)
self.background_tasks.add_task(self._notify_admin, name)
return {"id": user_id, "name": name}
def _send_welcome(self, email: str, name: str):
print(f"[BG] Welcome email sent to {email}")
def _notify_admin(self, name: str):
print(f"[BG] Admin notified: new user '{name}'")
def get_user_service(background_tasks: BackgroundTasks) -> UserService:
return UserService(background_tasks=background_tasks)
@app.post("/users")
def create_user(
name: str,
email: str,
service: UserService = Depends(get_user_service)
):
user = service.create_user(name, email)
return user
Output:
{"id": 1, "name": "Alice"}
Background output:
[BG] Welcome email sent to alice@example.com
[BG] Admin notified: new user 'Alice'
The route handler stays thin — it calls service.create_user() and returns the result. The service encapsulates both the business logic and the side effects. This pattern scales well as your service layer grows, because you can add more background tasks inside the service without touching the route function.
Using Async Functions as Background Tasks
FastAPI’s BackgroundTasks supports both regular (synchronous) and async functions. If your background work uses an async library — like aiofiles for async file I/O or httpx for async HTTP — you can pass an async function directly to add_task() and FastAPI will await it correctly inside the event loop.
# background_async.py
from fastapi import FastAPI, BackgroundTasks
import asyncio
app = FastAPI()
async def async_log(message: str):
# Simulates async I/O -- e.g. writing to an async database
await asyncio.sleep(0.5)
print(f"[ASYNC LOG] {message}")
async def async_notify(service: str, payload: dict):
await asyncio.sleep(0.2)
print(f"[ASYNC NOTIFY] {service}: {payload}")
@app.post("/event")
async def record_event(event_type: str, background_tasks: BackgroundTasks):
background_tasks.add_task(async_log, f"event={event_type}")
background_tasks.add_task(async_notify, "slack", {"event": event_type})
return {"received": True}
Output (client sees immediately):
{"received": true}
Background output (after response is sent):
[ASYNC NOTIFY] slack: {'event': 'page_view'}
[ASYNC LOG] event=page_view
Notice the order: the notify task has a shorter sleep (0.2s) so it finishes first, even though log was added first. When both tasks are async and run in the same event loop, they interleave based on their await points — not strictly in add order. If strict ordering matters, chain the calls inside a single wrapper function instead of adding them as separate tasks.
Error Handling in Background Tasks
Background tasks run outside the normal request-response cycle, which means unhandled exceptions in a task do not automatically produce an HTTP error response — the client already received their 200. This makes error handling in background tasks a discipline you have to build yourself.
The safest pattern is to wrap every background function body in a try/except and log or report the failure independently. Do not let exceptions propagate silently.
# background_errors.py
from fastapi import FastAPI, BackgroundTasks
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
def send_notification(user_id: int, channel: str):
try:
if channel not in ("email", "sms", "push"):
raise ValueError(f"Unknown channel: {channel}")
# Simulate sending
logger.info(f"Notification sent to user {user_id} via {channel}")
except Exception as exc:
# Log the failure -- do not re-raise (would silently kill remaining tasks)
logger.error(f"Notification failed for user {user_id}: {exc}")
def update_last_seen(user_id: int):
# This one always succeeds
logger.info(f"Updated last_seen for user {user_id}")
@app.post("/ping/{user_id}")
def ping_user(user_id: int, channel: str, background_tasks: BackgroundTasks):
background_tasks.add_task(send_notification, user_id, channel)
background_tasks.add_task(update_last_seen, user_id)
return {"pong": True}
Output (bad channel — client sees success, error is logged):
{"pong": true}
INFO: Notification failed for user 5: Unknown channel: carrier_pigeon
INFO: Updated last_seen for user 5
By catching the exception inside the task function, we ensure the second task (update_last_seen) still runs. If we had let the exception propagate out of send_notification, FastAPI would catch it and log a server error, but update_last_seen would be skipped. Always wrap background task bodies in try/except when you have multiple tasks queued.
Real-Life Example: User Registration with Background Side Effects
Let us build a realistic user registration endpoint that uses BackgroundTasks for all of its post-registration side effects: sending a welcome email, notifying an admin Slack channel, and logging the event to a file. The route itself stays fast — it creates the user record and returns immediately.
# registration_service.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel, EmailStr
import logging
import datetime
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
app = FastAPI()
# Simulated user store
users: dict = {}
class RegisterRequest(BaseModel):
name: str
email: str
plan: str = "free"
# ----- Background task functions -----
def send_welcome_email(email: str, name: str, plan: str):
try:
# Replace with your actual SMTP or email service call
logger.info(f"[EMAIL] Welcome email sent to {email} (plan={plan})")
except Exception as exc:
logger.error(f"[EMAIL] Failed for {email}: {exc}")
def notify_slack_admin(name: str, email: str, plan: str):
try:
# Replace with requests.post() to your Slack webhook
message = f"New signup: {name} ({email}) on {plan} plan"
logger.info(f"[SLACK] {message}")
except Exception as exc:
logger.error(f"[SLACK] Notification failed: {exc}")
def write_signup_log(user_id: int, email: str, plan: str):
try:
entry = {
"user_id": user_id,
"email": email,
"plan": plan,
"timestamp": datetime.datetime.utcnow().isoformat()
}
with open("signups.log", "a") as f:
f.write(str(entry) + "\n")
logger.info(f"[LOG] Signup logged for user {user_id}")
except Exception as exc:
logger.error(f"[LOG] Failed to write signup log: {exc}")
# ----- Route -----
@app.post("/register", status_code=201)
def register(payload: RegisterRequest, background_tasks: BackgroundTasks):
# Validate uniqueness
existing = next((u for u in users.values() if u["email"] == payload.email), None)
if existing:
raise HTTPException(status_code=409, detail="Email already registered")
# Create user record
user_id = len(users) + 1
users[user_id] = {
"id": user_id,
"name": payload.name,
"email": payload.email,
"plan": payload.plan,
}
# Queue all side effects -- they run AFTER the response is sent
background_tasks.add_task(send_welcome_email, payload.email, payload.name, payload.plan)
background_tasks.add_task(notify_slack_admin, payload.name, payload.email, payload.plan)
background_tasks.add_task(write_signup_log, user_id, payload.email, payload.plan)
return {"id": user_id, "name": payload.name, "plan": payload.plan}
Output (client sees instantly):
{"id": 1, "name": "Alice", "plan": "pro"}
Background output (logged after response):
2026-08-01 06:00:01 [INFO] [EMAIL] Welcome email sent to alice@example.com (plan=pro)
2026-08-01 06:00:01 [INFO] [SLACK] New signup: Alice (alice@example.com) on pro plan
2026-08-01 06:00:01 [INFO] [LOG] Signup logged for user 1
This pattern keeps the route handler focused on one job — creating the user record and returning the result. Every side effect is a named function with its own error handling. To extend it, add another background_tasks.add_task() call without touching the core registration logic. You can graduate individual tasks to Celery later if any of them become too slow, without refactoring the whole route.
Frequently Asked Questions
When should I use BackgroundTasks vs Celery?
Use BackgroundTasks when your task is fast (under a few seconds), does not need retries, does not need scheduling, and runs in the same process. Use Celery (or ARQ, or Dramatiq) when your task is slow (video processing, complex ML inference), needs guaranteed retry on failure, needs to run on a schedule, or needs to run across multiple worker processes. The rule of thumb: if you would not mind the task silently failing and never retrying, BackgroundTasks is fine. If failure has business consequences, use a proper task queue.
What happens if a background task raises an exception?
FastAPI catches the exception and logs a server error, but the HTTP response has already been sent to the client — they will not see a 500 error. Any background tasks queued after the failing task will be skipped for that request. This is why wrapping the body of every background function in a try/except is important: it lets subsequent tasks continue running even when one fails, and ensures failures are logged where you can see them.
Can I mix async and sync background task functions?
Yes. BackgroundTasks.add_task() accepts both synchronous and async functions. FastAPI detects whether the function is a coroutine and either awaits it (async) or calls it directly (sync). The important caveat for synchronous tasks: if they do heavy CPU work (not I/O), they will block the event loop while running. For CPU-heavy background work, use asyncio.run_in_executor() to offload to a thread pool, or switch to a real task queue.
How do I use a database session inside a background task?
Do not reuse the request-scoped database session from the route handler. By the time the background task runs, that session may have been closed. Instead, open a fresh database session inside the background task function itself. If you use SQLAlchemy with FastAPI, create a new SessionLocal() at the top of the task, use it, and close it in a finally block. The same principle applies to any request-scoped resource.
How do I test routes that use BackgroundTasks?
Use FastAPI’s TestClient. Background tasks run synchronously during tests when you use TestClient — they execute before the response is returned, which makes them easy to assert on. Simply call your route via client.post(), then check any side effects (files written, database rows created, mock functions called) immediately after. No special setup is needed for basic cases.
Is there a limit to how many background tasks I can add per request?
There is no hard limit enforced by FastAPI — you can call add_task() as many times as you need. The practical limit is your server’s memory and the latency implications for the next request that arrives. All queued tasks from a request run sequentially in the same worker, so a flood of slow tasks can delay other requests. If you are regularly adding more than two or three tasks per request, that is a signal to evaluate whether a real task queue would serve you better.
Conclusion
FastAPI’s BackgroundTasks is one of those features that solves a common problem so cleanly that you wonder why you ever did it any other way. You inject BackgroundTasks into a route, call add_task() with your function and its arguments, and FastAPI takes care of the rest — running your function after the response is sent, with full support for both sync and async code. The real-life registration example above shows how to structure this for production: named task functions, individual error handling, and a route handler that stays focused on its core job.
The next step is to extend the registration example with a real email library like fastapi-mail or sendgrid, and replace the print-based Slack notification with an actual webhook call. Both are drop-in replacements for the stub functions we wrote — the background task pattern stays identical. When any individual task eventually outgrows BackgroundTasks (because it gets slow, needs retries, or needs scheduling), you can migrate just that task to Celery without refactoring the rest of the route.
For official documentation and deeper reading, see the FastAPI Background Tasks docs and the Starlette BackgroundTask reference.