Background Tasks in FastAPI

BackgroundTasks — execute tasks after the response has been sent to the client.

Basic Usage

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # Runs after the response is sent
    print(f"Sending email to {email}: {message}")

@app.post("/tasks/")
def create_task(task: TaskCreate, background_tasks: BackgroundTasks):
    # Create the task
    db_task = save_task(task)

    # Add a background task
    background_tasks.add_task(send_email, "user@example.com", f"Task created: {task.title}")

    # Response is sent immediately, email is sent afterwards
    return db_task

Multiple Tasks

@app.post("/users/register")
def register(user: UserCreate, background_tasks: BackgroundTasks):
    db_user = create_user(user)

    background_tasks.add_task(send_welcome_email, db_user.email)
    background_tasks.add_task(notify_admin, db_user.id)
    background_tasks.add_task(update_stats)

    return {"message": "Registered"}

Via Depends

def get_background_tasks(background_tasks: BackgroundTasks):
    return background_tasks

@app.post("/tasks/")
def create_task(
    task: TaskCreate,
    bg: BackgroundTasks = Depends(),
):
    db_task = save_task(task)
    bg.add_task(process_task, db_task.id)
    return db_task

Limitations

• Runs in the same process — will block if the task takes too long
• No retry on errors
• No monitoring

When to Use

• Sending email/SMS after an action
• Updating a cache
• Logging to a third-party service
• Short-lived tasks (< 30 sec)

For Heavy Tasks — Celery or ARQ

pip install celery redis

@celery.task
def heavy_processing(task_id: int):
    ...  # long processing in a separate worker

@app.post("/tasks/")
def create_task(task: TaskCreate):
    db_task = save_task(task)
    heavy_processing.delay(db_task.id)  # non-blocking
    return db_task