FastAPI Background Task Fix
In this tutorial, you'll learn about FastAPI Background Task Fix. We cover key concepts, practical examples, and best practices.
The Problem
Sending emails, processing uploads, or generating reports in the request handler blocks the response. The user waits for the email to send before seeing the confirmation page.
Quick Fix
Wrong — blocking the response
@app.post("/register")
async def register(user: UserCreate):
db_user = await create_user(user)
await send_welcome_email(db_user.email) # User waits for this
return {"message": "User created"}
Output: Response takes 2-5 seconds while the email sends. The user sees a loading spinner.
Correct — BackgroundTasks
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def send_welcome_email(email: str):
# Sync function runs in thread pool
smtp = smtplib.SMTP("smtp.example.com")
smtp.send_message(...)
smtp.quit()
@app.post("/register")
async def register(user: UserCreate, tasks: BackgroundTasks):
db_user = await create_user(user)
tasks.add_task(send_welcome_email, db_user.email)
return {"message": "User created"}
Output: Response returns immediately. Email sends in the background.
Async background tasks
async def process_image_async(image_id: int):
async with aiohttp.ClientSession() as session:
async with session.post("https://api.process.com", json={"id": image_id}) as resp:
result = await resp.json()
await save_result(image_id, result)
@app.post("/upload")
async def upload(image: UploadFile, tasks: BackgroundTasks):
image_id = await save_image(image)
tasks.add_task(process_image_async, image_id)
return {"image_id": image_id}
Multiple background tasks
@app.post("/order")
async def create_order(order: OrderCreate, tasks: BackgroundTasks):
order_id = await save_order(order)
tasks.add_task(send_confirmation_email, order_id)
tasks.add_task(update_inventory, order.items)
tasks.add_task(notify_admin, order_id)
return {"order_id": order_id}
Dependencies with background tasks
def get_db_session():
db = Database()
yield db
db.close()
def log_activity(db: Database = Depends(get_db_session)):
def _log(user_id: int, action: str):
db.execute("INSERT INTO logs VALUES (?, ?)", (user_id, action))
return _log
@app.post("/login")
async def login(tasks: BackgroundTasks, log: BackgroundTask = Depends(log_activity)):
tasks.add_task(log, user_id=1, action="login")
return {"status": "ok"}
Error handling in background tasks
import logging
def send_email(email: str):
try:
smtp = smtplib.SMTP("smtp.example.com")
smtp.sendmail(...)
smtp.quit()
except Exception as e:
logging.error(f"Failed to send email to {email}: {e}")
Prevention
- Use
BackgroundTasksfor non-critical post-response work. - Keep background tasks stateless — they can't return data to the client.
- Handle exceptions inside background tasks — unhandled errors are silently lost.
Common Mistakes with background task
- Misunderstanding that
Stringis[Char]with poor performance for large text operations - Using
foldlinstead offoldl'causing stack overflow on large lists - Forgetting
deriving (Show, Eq)on custom data types needed for debugging
These mistakes appear frequently in real-world FASTAPI code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.
Practice Exercise
Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.
This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro