Background Tasks in FastAPI

Not everything should happen before the response. Sending emails, processing uploads, calling webhooks, and updating analytics are all work that can happen after you return a 200 to the user. FastAPI gives you two paths: built-in BackgroundTasks for simple in-process work, and external task queues like Celery for anything heavier.

FastAPI BackgroundTasks

BackgroundTasks runs functions after the response is sent, in the same process.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_welcome_email(email: str, name: str):
    # This runs after the response is sent
    print(f"Sending welcome email to {email}")
    # In practice: call your email service here
    import smtplib
    msg = f"Welcome, {name}! Thanks for signing up."
    # ... send the email

@app.post("/users")
async def create_user(
    email: str,
    name: str,
    background_tasks: BackgroundTasks,
):
    user = await save_user(email, name)
    
    # Queue the email to run after the response
    background_tasks.add_task(send_welcome_email, email, name)
    
    return {"id": user.id, "email": email}

The response returns immediately. The email sends in the background. If the email fails, the user still got their 200 response.

Multiple background tasks

You can queue multiple tasks. They run in order.

@app.post("/orders")
async def create_order(
    order: OrderCreate,
    background_tasks: BackgroundTasks,
):
    saved_order = await save_order(order)
    
    # All three run after the response, in order
    background_tasks.add_task(send_order_confirmation, saved_order)
    background_tasks.add_task(notify_warehouse, saved_order)
    background_tasks.add_task(update_analytics, "order_created", saved_order.id)
    
    return saved_order

BackgroundTasks in dependencies

You can add background tasks from dependencies, not just route handlers.

from fastapi import Depends, BackgroundTasks

def log_request(background_tasks: BackgroundTasks):
    # This dependency adds a background task
    background_tasks.add_task(
        write_audit_log, "request processed"
    )

@app.get("/items", dependencies=[Depends(log_request)])
async def list_items():
    return await get_items()

Async background tasks

Background tasks can be async functions.

async def process_upload(file_path: str, user_id: str):
    async with aiofiles.open(file_path, "rb") as f:
        data = await f.read()
    
    # Process the file
    result = await analyze_image(data)
    
    # Store results
    await db.uploads.update(
        {"user_id": user_id, "file_path": file_path},
        {"status": "processed", "result": result},
    )

@app.post("/upload")
async def upload_file(
    file: UploadFile,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
):
    # Save the file quickly
    file_path = f"/tmp/uploads/{file.filename}"
    async with aiofiles.open(file_path, "wb") as f:
        content = await file.read()
        await f.write(content)
    
    # Process in background
    background_tasks.add_task(process_upload, file_path, user["id"])
    
    return {"status": "uploaded", "processing": True}

Limitations of BackgroundTasks

BackgroundTasks are in-process. They have real limitations:

• No retries. If a task fails, it fails silently. No automatic retry.
• No persistence. If the server crashes, queued tasks are lost.
• No distribution. Tasks run on the same server that handled the request.
• Blocks the event loop if the task is synchronous and long-running (FastAPI runs sync background tasks in a thread pool, but still).
• No monitoring. No built-in way to track task status or failures.

Use BackgroundTasks for: sending emails, writing logs, lightweight webhooks, cache warming. For anything that must not be lost, use a real task queue.

Celery for distributed tasks

Celery is the standard Python library for distributed task queues. Tasks are serialized, sent to a message broker (Redis or RabbitMQ), and executed by worker processes.

Setup

# celery_app.py
from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,  # Re-deliver if worker crashes
    worker_prefetch_multiplier=1,
)

Define tasks

# tasks.py
from celery_app import celery_app
import time

@celery_app.task(bind=True, max_retries=3)
def send_email(self, to: str, subject: str, body: str):
    try:
        # Call email service
        response = email_client.send(to=to, subject=subject, body=body)
        return {"status": "sent", "message_id": response.id}
    except ConnectionError as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

@celery_app.task(bind=True, max_retries=5)
def process_video(self, video_id: str):
    self.update_state(state="PROCESSING", meta={"progress": 0})
    
    video = download_video(video_id)
    self.update_state(state="PROCESSING", meta={"progress": 25})
    
    transcoded = transcode(video, format="mp4", quality="720p")
    self.update_state(state="PROCESSING", meta={"progress": 75})
    
    upload_to_cdn(transcoded)
    self.update_state(state="PROCESSING", meta={"progress": 100})
    
    return {"status": "completed", "url": f"https://cdn.example.com/{video_id}.mp4"}

@celery_app.task
def generate_report(user_id: str, report_type: str):
    data = fetch_report_data(user_id, report_type)
    pdf = render_pdf(data)
    url = upload_to_s3(pdf)
    notify_user(user_id, f"Your report is ready: {url}")
    return {"url": url}

Call from FastAPI

from fastapi import FastAPI
from tasks import send_email, process_video, generate_report

app = FastAPI()

@app.post("/users")
async def create_user(data: UserCreate):
    user = await save_user(data)
    
    # Dispatch to Celery (returns immediately)
    task = send_email.delay(
        to=user.email,
        subject="Welcome!",
        body=f"Hi {user.name}, welcome aboard.",
    )
    
    return {"id": user.id, "email_task_id": task.id}

@app.post("/videos/{video_id}/process")
async def start_processing(video_id: str):
    task = process_video.delay(video_id)
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    from celery.result import AsyncResult
    result = AsyncResult(task_id)
    
    response = {
        "task_id": task_id,
        "status": result.status,
    }
    
    if result.status == "PROCESSING":
        response["progress"] = result.info.get("progress", 0)
    elif result.status == "SUCCESS":
        response["result"] = result.result
    elif result.status == "FAILURE":
        response["error"] = str(result.result)
    
    return response

Run workers

# Start a worker
celery -A celery_app worker --loglevel=info --concurrency=4

# Start with specific queues
celery -A celery_app worker -Q emails,reports --loglevel=info

# Start the beat scheduler for periodic tasks
celery -A celery_app beat --loglevel=info

Periodic tasks with Celery Beat

from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "tasks.cleanup_sessions",
        "schedule": crontab(minute=0, hour="*/2"),  # Every 2 hours
    },
    "generate-daily-report": {
        "task": "tasks.daily_report",
        "schedule": crontab(minute=0, hour=6),  # 6 AM daily
    },
    "health-check-services": {
        "task": "tasks.health_check",
        "schedule": 300.0,  # Every 5 minutes
    },
}

When to use what

Scenario	Approach
Send a confirmation email	BackgroundTasks
Write an audit log entry	BackgroundTasks
Process a video upload	Celery
Generate a PDF report	Celery
Sync data with third-party API	Celery
Update a cache	BackgroundTasks
Run ML inference on uploaded image	Celery
Webhook notification (non-critical)	BackgroundTasks
Webhook notification (must deliver)	Celery with retries
Scheduled cleanup job	Celery Beat

The rule of thumb: if you can afford to lose the task on a server restart, use BackgroundTasks. If the task must complete, must retry on failure, or takes more than a few seconds, use Celery.

Error handling patterns

BackgroundTasks error handling

Since BackgroundTasks do not report errors, wrap them yourself.

import logging
import traceback

logger = logging.getLogger(__name__)

def safe_background_task(func):
    async def wrapper(*args, **kwargs):
        try:
            if asyncio.iscoroutinefunction(func):
                await func(*args, **kwargs)
            else:
                func(*args, **kwargs)
        except Exception:
            logger.error(
                f"Background task {func.__name__} failed:\n"
                f"{traceback.format_exc()}"
            )
    return wrapper

@app.post("/notify")
async def notify(
    message: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(
        safe_background_task(send_notification), message
    )
    return {"status": "queued"}

Celery error handling with dead letter queue

@celery_app.task(
    bind=True,
    max_retries=3,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
)
def resilient_task(self, data):
    try:
        return process(data)
    except PermanentError:
        # Don't retry, log and move on
        logger.error(f"Permanent failure processing {data}")
        return {"status": "failed", "reason": "permanent_error"}

Background tasks are about matching the right tool to the right problem. Start with FastAPI’s built-in BackgroundTasks for simplicity. Graduate to Celery when you need reliability, retries, distribution, or monitoring. The transition is straightforward because both approaches dispatch work the same way from your route handlers.