Background Tasks in FastAPI

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

@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"}

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

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