Background Tasks в FastAPI

BackgroundTasks — выполнение задач после отправки ответа клиенту.

Базовое использование

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # Выполняется после ответа
    print(f"Отправка письма на {email}: {message}")

@app.post("/tasks/")
def create_task(task: TaskCreate, background_tasks: BackgroundTasks):
    # Создать задачу
    db_task = save_task(task)

    # Добавить фоновую задачу
    background_tasks.add_task(send_email, "user@example.com", f"Создана задача: {task.title}")

    # Ответ отправляется немедленно, email — потом
    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": "Зарегистрирован"}

Через 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

Ограничения

• Выполняется в том же процессе — блокирует, если задача долгая
• Нет retry при ошибках
• Нет мониторинга

Когда использовать

• Отправка email/SMS после действия
• Обновление кэша
• Логирование в сторонний сервис
• Быстрые задачи (< 30 сек)

Для тяжёлых задач — Celery или ARQ

pip install celery redis

@celery.task
def heavy_processing(task_id: int):
    ...  # долгая обработка в отдельном воркере

@app.post("/tasks/")
def create_task(task: TaskCreate):
    db_task = save_task(task)
    heavy_processing.delay(db_task.id)  # не блокирует
    return db_task