Overview
Background tasks allow you to execute functions after returning an HTTP response to the client. This is useful for operations that don’t need to complete before sending the response, such as sending emails, logging, or processing data.
The tasks are executed asynchronously after the response is sent, ensuring fast response times while still completing necessary work.
BackgroundTasks class
The BackgroundTasks class is used to queue functions that will run in the background after the response is returned.
Usage
Add BackgroundTasks as a parameter to your route handler, and FastrAPI will automatically inject it:
from fastrapi import FastrAPI, BackgroundTasks
app = FastrAPI()
@app.post("/send-notification")
def send_notification(background_tasks: BackgroundTasks):
# Add a task to run in the background
background_tasks.add_task(send_email, "[email protected]")
# Response is returned immediately
return {"status": "notification queued"}
Methods
add_task()
Adds a function to be executed in the background after the response is sent.
The function to execute in the background
Positional arguments to pass to the function
Returns nothing. The task is queued for execution.
def write_log(message: str, level: str):
with open("app.log", "a") as f:
f.write(f"[{level}] {message}\n")
@app.post("/process")
def process_data(background_tasks: BackgroundTasks):
# Queue the background task with arguments
background_tasks.add_task(write_log, "Processing started", "INFO")
return {"status": "processing"}
Examples
Send email after registration
from fastrapi import FastrAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr
app = FastrAPI()
class User(BaseModel):
email: EmailStr
name: str
def send_welcome_email(email: str, name: str):
# Simulate sending an email
print(f"Sending welcome email to {email}")
# Email sending logic here
time.sleep(2) # Simulated delay
print(f"Email sent to {name}")
@app.post("/register")
def register_user(user: User, background_tasks: BackgroundTasks):
# Add email task to background
background_tasks.add_task(send_welcome_email, user.email, user.name)
# Return immediately without waiting for email
return {"message": "Registration successful", "email": user.email}
Multiple background tasks
You can add multiple tasks that will all execute after the response:
def log_action(action: str):
print(f"Action logged: {action}")
def update_analytics(user_id: int):
print(f"Analytics updated for user {user_id}")
def send_notification(user_id: int):
print(f"Notification sent to user {user_id}")
@app.post("/action/{user_id}")
def perform_action(user_id: int, background_tasks: BackgroundTasks):
# Add multiple background tasks
background_tasks.add_task(log_action, "user_action")
background_tasks.add_task(update_analytics, user_id)
background_tasks.add_task(send_notification, user_id)
# All tasks execute after this response
return {"status": "action completed"}
File processing
from fastrapi import FastrAPI, BackgroundTasks
import os
app = FastrAPI()
def process_file(filename: str):
# Simulate heavy processing
print(f"Processing {filename}...")
time.sleep(5)
print(f"Finished processing {filename}")
# Clean up
os.remove(filename)
@app.post("/upload")
def upload_file(file_content: str, background_tasks: BackgroundTasks):
filename = "temp_file.txt"
# Save the file
with open(filename, "w") as f:
f.write(file_content)
# Process in background
background_tasks.add_task(process_file, filename)
return {"message": "File uploaded and queued for processing"}
Database cleanup
def cleanup_expired_sessions():
# Clean up expired database records
print("Cleaning up expired sessions...")
# Database cleanup logic
@app.post("/logout")
def logout(user_id: int, background_tasks: BackgroundTasks):
# Perform immediate logout
# Then clean up expired sessions in background
background_tasks.add_task(cleanup_expired_sessions)
return {"message": "Logged out successfully"}
Best practices
Keep tasks lightweight
Background tasks run after the response is sent but still consume server resources. For very heavy operations, consider using a dedicated task queue like Celery.
# Good: Quick operations
background_tasks.add_task(send_email, user.email)
background_tasks.add_task(log_action, "user_login")
# Consider a task queue instead: Very heavy operations
# background_tasks.add_task(process_million_records) # Too heavy!
Error handling
Errors in background tasks are logged but don’t affect the response (which has already been sent). Make sure your background functions handle errors gracefully:
def send_email_safe(email: str):
try:
# Email sending logic
send_email(email)
except Exception as e:
print(f"Failed to send email: {e}")
# Log error, retry, or handle appropriately
@app.post("/notify")
def notify_user(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(send_email_safe, email)
return {"status": "notification queued"}
Task execution order
Tasks are executed in the order they were added:
@app.post("/ordered-tasks")
def ordered_tasks(background_tasks: BackgroundTasks):
background_tasks.add_task(task1) # Executes first
background_tasks.add_task(task2) # Executes second
background_tasks.add_task(task3) # Executes third
return {"status": "tasks queued"}