How to Use FastAPI Background Tasks with Simple Examples

When you first hear “background tasks” in a web framework, the mind jumps to queues, workers, and a lot of moving parts. FastAPI, however, keeps things surprisingly lightweight. You can off‑load a short‑lived job—sending an email, cleaning a cache, logging a request—without pulling in Celery or RabbitMQ. The following guide walks through the core ideas, then shows three practical examples you can copy‑paste into a fresh project.

Why Background Tasks Matter in FastAPI

FastAPI processes each request in an asynchronous fashion. If you block that flow with a time‑consuming operation, the client waits, and the server’s throughput drops. A background task runs after the response is sent, letting the client move on while the server finishes the work quietly in the same process.

• Speed up response time without sacrificing functionality.
• Simplify deployment – no external broker needed for short jobs.
• Stay async‑friendly – tasks cooperate with FastAPI’s event loop.

Getting Started: The Bare Minimum

First, install FastAPI and an ASGI server, typically uvicorn:

pip install fastapi uvicorn

Then create main.py with a single endpoint that schedules a background function:

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message + "\n")
@app.post("/notify/")
async def notify(user: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"User {user} sent a notification")
return {"detail": "Notification received"}

Run it with uvicorn main:app --reload. When you POST to /notify/, the response returns instantly, while write_log appends to log.txt in the background.

Example 1: Sending an Email After a Signup

Imagine a simple signup flow where you want to welcome the new user by email. Using the built‑in email.message library (or any SMTP wrapper) works just fine for a demonstration.

import smtplib
from email.message import EmailMessage
def send_welcome_email(to_address: str):
msg = EmailMessage()
msg["Subject"] = "Welcome aboard!"
msg["From"] = "no-reply@example.com"
msg["To"] = to_address
msg.set_content("Thanks for joining us. We’re excited to have you!")
# This example uses localhost; replace with real credentials as needed
with smtplib.SMTP("localhost") as smtp:
smtp.send_message(msg)

Hook it into an endpoint:

@app.post("/signup/")
async def signup(email: str, background_tasks: BackgroundTasks):
# (Pretend we store the user here)
background_tasks.add_task(send_welcome_email, email)
return {"detail": "Signup successful – check your inbox"}

Because the email is sent after the JSON reply, the user never notices the extra latency.

Example 2: Generating a Thumbnail Asynchronously

Image processing can be CPU‑heavy. Let’s spin off a Pillow routine that resizes an uploaded picture and saves a thumbnail.

from PIL import Image
import os
def create_thumbnail(file_path: str):
size = (128, 128)
with Image.open(file_path) as img:
img.thumbnail(size)
thumb_path = f"{os.path.splitext(file_path)[0]}_thumb.jpg"
img.save(thumb_path, "JPEG")

Endpoint that accepts an upload:

@app.post("/upload/")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
file_location = f"uploads/{file.filename}"
with open(file_location, "wb") as buffer:
buffer.write(await file.read())
background_tasks.add_task(create_thumbnail, file_location)
return {"detail": f"{file.filename} received, thumbnail on its way"}

The client sees a success message right away, while the thumbnail generation happens quietly.

Example 3: Cleaning Up Stale Sessions Periodically

Sometimes you need a recurring clean‑up, but you don’t want a full‑blown scheduler. FastAPI’s background tasks can be combined with asyncio.sleep to create a simple loop that runs as long as the server is alive.

import asyncio
from datetime import datetime, timedelta
async def session_cleaner():
while True:
# Placeholder: find sessions older than 30 minutes and delete them
print(f"[{datetime.utcnow()}] Cleaning stale sessions...")
await asyncio.sleep(1800)  # run every 30 minutes
@app.on_event("startup")
async def start_cleaner():
asyncio.create_task(session_cleaner())

When the app boots, session_cleaner launches in the background and keeps the database tidy without any external cron job.

Best Practices and Gotchas

• Keep tasks short. If a job might exceed a few seconds, consider a dedicated worker queue instead.
• Avoid blocking calls. Use async‑compatible libraries (e.g., aiosmtplib for email) when possible.
• Handle exceptions. Background tasks run silently; wrap code in try/except and log failures.
• Watch file descriptors. Opening files without closing them can leak resources; use with blocks.

When to Reach for a Full Queue System

If you find yourself queuing long‑running data pipelines, heavy video transcoding, or tasks that need retries after failure, FastAPI’s built‑in background feature will feel cramped. In those cases, integrating Celery, RQ, or a cloud‑based task service gives you durability, monitoring, and scaling out of the box.

For the everyday use‑case—sending a confirmation email, writing a log line, resizing a picture—FastAPI’s BackgroundTasks class provides a neat, no‑frills solution that stays within the same process.