Что такое фоновые задачи (background tasks) в FastAPI и когда их следует использовать?

Базовый пример

import logging
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr

app = FastAPI()
log = logging.getLogger(__name__)

def send_welcome_email(email: str, username: str) -> None:
    # Синхронная функция — FastAPI выполнит её в ThreadPoolExecutor
    log.info("Sending welcome email to %s", email)
    # ... вызов SMTP или Resend API ...

class UserCreate(BaseModel):
    email: EmailStr
    username: str

@app.post("/users", status_code=201)
async def create_user(
    body: UserCreate,
    background_tasks: BackgroundTasks,
):
    # Сохраняем пользователя в БД (здесь опущено для краткости)
    background_tasks.add_task(send_welcome_email, body.email, body.username)
    return {"id": 42, "username": body.username}

Механизм работы

BackgroundTasks — класс из fastapi, реализованный поверх starlette.background.BackgroundTasks. FastAPI внедряет его через DI: достаточно добавить параметр типа BackgroundTasks в сигнатуру обработчика. Задача добавляется методом background_tasks.add_task(func, *args, **kwargs) и выполняется после того, как ответ отправлен клиенту — в том же процессе и том же event loop (для async-функций) или в пуле потоков (для sync-функций).

Async-вариант

import httpx
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def notify_webhook(url: str, payload: dict) -> None:
    async with httpx.AsyncClient() as client:
        await client.post(url, json=payload, timeout=10.0)

@app.post("/events")
async def publish_event(
    payload: dict,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(notify_webhook, "https://hooks.example.com", payload)
    return {"status": "accepted"}

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

• Отправка email или push-уведомления после регистрации/заказа.
• Запись в audit log, когда задержка ответа нежелательна.
• Инвалидация кэша или обновление счётчиков, не требующее транзакции.
• Лёгкие webhook-уведомления во внешние системы.

Когда BackgroundTasks недостаточно

BackgroundTasks выполняется в том же процессе. Если процесс завершится (crash, deploy), задача будет потеряна. Для надёжной доставки, повторных попыток и мониторинга используйте очередь задач: ARQ (asyncio + Redis), Celery (с broker Redis или RabbitMQ), RQ.

# Пример с ARQ вместо BackgroundTasks для надёжной доставки
import arq
from fastapi import FastAPI

app = FastAPI()

@app.post("/orders", status_code=201)
async def create_order(body: dict, redis: arq.ArqRedis):
    order_id = 99  # сохранить в БД
    await redis.enqueue_job("send_order_confirmation", order_id=order_id)
    return {"id": order_id}

Подводные камни

• Потеря задачи при сбое процесса. BackgroundTasks не персистентны — если Uvicorn упадёт, задача не выполнится и не будет повторена.
• Исключения не доходят до клиента. Ошибка в фоновой задаче логируется, но не влияет на HTTP-ответ; нужно явно настроить обработку ошибок (try/except + logging).
• Shared state и DB-сессии. Нельзя использовать SQLAlchemy-сессию, созданную в обработчике, внутри фоновой задачи — она может быть закрыта к моменту выполнения. Создавайте новую сессию внутри задачи.
• Нет контроля конкурентности. При высокой нагрузке множество фоновых задач выполняются одновременно без ограничений; для ограничения используйте Semaphore или выносите задачи в worker.
• Нет retry-логики. BackgroundTasks не поддерживает повторные попытки при ошибке сети или SMTP; реализуйте её вручную или переходите на ARQ/Celery.
• Тестирование. В тестах с TestClient фоновые задачи выполняются синхронно до возврата из вызова; с AsyncClient (httpx) — тоже, но это поведение не задокументировано и может измениться.
• Зависимости недоступны. Параметры с Depends() не передаются в функцию задачи автоматически — нужно явно передавать нужные объекты через аргументы add_task.

• Описывать background tasks только как термин и не показывать механизм на минимальном примере.
• Игнорировать ошибки, пустые данные, конкурентный доступ или границы транзакции.
• Не связывать поведение с официальным контрактом FastAPI и реальной эксплуатацией.

• Объясняет background tasks через последовательность действий, а не через набор ключевых слов.
• Приводит короткий кодовый пример или production-сценарий с ожидаемым поведением.
• Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.