FastAPI Background Tasks
Run work after the response with BackgroundTasks, and when to reach for Celery instead
Knowledge Debt detected
You can study this freely — but your score may plateau if these foundations have gaps. The Mastery badge requires them to be solid.
Explanation
BackgroundTasks lets you run code *after* the HTTP response is sent — ideal for fire-and-forget work like sending emails, logging analytics, or cleaning up files.
Basic usage:
python from fastapi import BackgroundTasks, FastAPI app = FastAPI() def send_welcome_email(email: str): # This runs after the response — no need to await time.sleep(2) # simulate slow email API print(f"Email sent to {email}") @app.post("/signup") def signup(email: str, background_tasks: BackgroundTasks): create_user(email) # runs now background_tasks.add_task(send_welcome_email, email) # runs after response return {"message": "Account created!"} # ← user gets this immediately FastAPI injects BackgroundTasks` by type annotation — no import or setup needed beyond the parameter.
Multiple tasks:
python @app.post("/order/{order_id}/complete") def complete_order(order_id: int, background_tasks: BackgroundTasks): mark_order_complete(order_id) background_tasks.add_task(send_confirmation_email, order_id) background_tasks.add_task(update_inventory, order_id) background_tasks.add_task(log_analytics_event, "order_completed", order_id) return {"status": "complete"} # All three tasks run concurrently after this response
Passing background tasks through dependencies:
python def log_request(background_tasks: BackgroundTasks, request: Request): background_tasks.add_task(write_log, request.url.path) @app.get("/items") def get_items(background_tasks: BackgroundTasks, _: None = Depends(log_request)): return fetch_items()
BackgroundTasks vs Celery — when to use each:
| | BackgroundTasks | Celery | |---|---|---| | Setup | Zero (built-in) | Redis/RabbitMQ + worker process | | Runs in | Same process as the web server | Separate worker processes | | Survives server crash | ❌ No | ✓ Yes (persisted queue) | | Retries on failure | ❌ No | ✓ Yes (configurable) | | Scheduling | ❌ No | ✓ Yes (celery beat) | | Use when | Simple, non-critical tasks | Critical, retryable, or scheduled tasks |
Rule: BackgroundTasks for email confirmations and analytics. Celery for payment processing, invoice generation, or anything that must not be lost.
Examples
Resizing an image after upload
The user gets an immediate response; the slow thumbnail operation happens after without blocking
import shutil
from pathlib import Path
from fastapi import BackgroundTasks, UploadFile
def create_thumbnail(path: str):
# CPU-bound — runs after response is sent
from PIL import Image
img = Image.open(path)
img.thumbnail((200, 200))
img.save(path.replace(".jpg", "_thumb.jpg"))
@app.post("/upload")
async def upload_image(file: UploadFile, background_tasks: BackgroundTasks):
dest = f"uploads/{file.filename}"
with open(dest, "wb") as f:
shutil.copyfileobj(file.file, f)
background_tasks.add_task(create_thumbnail, dest)
return {"filename": file.filename, "message": "Upload complete, thumbnail generating"}BackgroundTasks in a dependency
Wrapping BackgroundTasks in a dependency keeps audit logging out of route logic
from fastapi import BackgroundTasks, Depends, Request
import logging
def audit_log(action: str, user_id: int):
logging.info(f"User {user_id} performed: {action}")
class AuditLogger:
def __init__(self, background_tasks: BackgroundTasks):
self.bg = background_tasks
def log(self, action: str, user_id: int):
self.bg.add_task(audit_log, action, user_id)
@app.delete("/users/{user_id}")
def delete_user(
user_id: int,
current_user: User = Depends(get_current_user),
audit: AuditLogger = Depends(),
):
db.delete(user_id)
audit.log("delete_user", current_user.id)
return {"deleted": user_id}How well did you understand this?
Next in FastAPI
FastAPI + SQLAlchemy