Что такое BackgroundTasks в FastAPI?

BackgroundTasks в FastAPI

BackgroundTasks — это встроенный механизм FastAPI для выполнения задач в фоне, не блокируя ответ клиенту. Это полезно для асинхронных операций, которые не нужно выполнять сразу.

Основной синтаксис

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def write_log(message: str):
    time.sleep(1)  # Имитация долгой операции
    print(f"Логирование: {message}")

@app.get("/send-email/")
async def send_email(background_tasks: BackgroundTasks):
    # Добавляем задачу в фоновую очередь
    background_tasks.add_task(write_log, "Email отправлен")
    
    # Сразу возвращаем ответ клиенту
    return {"message": "Email будет отправлен в фоне"}

Порядок выполнения:

1. Клиент отправляет запрос
2. Функция обработчика выполняется
3. Ответ отправляется клиенту НЕМЕДЛЕННО
4. Фоновая задача выполняется ПОСЛЕ отправки ответа

Преимущества

• Быстрый ответ — клиент не ждет завершения задачи
• Простота — встроено в FastAPI, не нужно Celery или Redis
• Интеграция — легко использовать в обработчиках

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

✅ Используй BackgroundTasks для:

• Отправки email
• Логирования
• Обработки изображений (не критично по скорости)
• Очистки кэша
• Уведомлений

❌ Не используй для:

• Критических операций (транзакции в БД)
• Долгих операций (больше минуты)
• Операций, требующих гарантии выполнения
• Высоконагруженных систем

Примеры использования

1. Отправка email с параметрами

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class EmailRequest(BaseModel):
    email: str
    message: str

def send_email(email: str, message: str):
    print(f"Отправляю письмо на {email}: {message}")
    # В реальности: используй smtplib или другую библиотеку

@app.post("/send-message/")
async def send_message(request: EmailRequest, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, request.email, request.message)
    return {"status": "Message queued for sending"}

2. Несколько фоновых задач

@app.post("/process-order/")
async def process_order(order_id: int, background_tasks: BackgroundTasks):
    # Добавляем несколько задач
    background_tasks.add_task(send_email, f"order-{order_id}@company.com", "Order received")
    background_tasks.add_task(log_to_database, order_id)
    background_tasks.add_task(update_inventory, order_id)
    
    return {"status": "Order processing started"}

def send_email(email, message):
    print(f"Email to {email}: {message}")

def log_to_database(order_id):
    print(f"Логирование заказа {order_id}")

def update_inventory(order_id):
    print(f"Обновление инвентаря для заказа {order_id}")

3. С обработкой файлов

from fastapi import FastAPI, UploadFile
import os

app = FastAPI()

def process_image(filename: str):
    # Имитация обработки: resize, watermark и т.д.
    print(f"Обработка изображения: {filename}")
    # После обработки: удалить временный файл
    if os.path.exists(f"temp/{filename}"):
        os.remove(f"temp/{filename}")

@app.post("/upload-image/")
async def upload_image(file: UploadFile, background_tasks: BackgroundTasks):
    # Сохраняем файл
    with open(f"temp/{file.filename}", "wb") as f:
        f.write(await file.read())
    
    # Добавляем обработку в фон
    background_tasks.add_task(process_image, file.filename)
    
    return {"filename": file.filename, "status": "Processing"}

Важные моменты

1. BackgroundTasks выполняются в ОДНОМ потоке

Если одна задача зависает, остальные ждут.

2. Нет гарантии выполнения

Если приложение упадет, фоновая задача может не выполниться. Для критических операций используй Celery или RQ.

3. Контекст недоступен

# ❌ Это НЕ работает - контекст запроса уже закончился
@app.get("/example/")
async def example(request: Request, background_tasks: BackgroundTasks):
    def task():
        print(request.client.host)  # Может не работать!
    background_tasks.add_task(task)

# ✅ Правильно - передай нужные данные
@app.get("/example/")
async def example(request: Request, background_tasks: BackgroundTasks):
    client_host = request.client.host
    background_tasks.add_task(log_request, client_host)

BackgroundTasks vs другие подходы

# 1. BackgroundTasks — простые некритические задачи
background_tasks.add_task(send_email, email)

# 2. Celery — сложные, долгие, критические задачи
send_email_task.delay(email)

# 3. asyncio.create_task() — асинхронные задачи
asyncio.create_task(send_email_async(email))

Реальный пример: Регистрация пользователя

from fastapi import FastAPI, BackgroundTasks
from sqlalchemy import Session
from database import get_db

app = FastAPI()

def send_welcome_email(email: str):
    print(f"Отправляю приветственное письмо на {email}")
    # Используй email провайдер (SendGrid, AWS SES и т.д.)

def log_user_registration(user_id: int, db: Session):
    # Логируем регистрацию
    print(f"Пользователь {user_id} зарегистрирован")

@app.post("/register/")
async def register_user(
    email: str,
    name: str,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    # Сохраняем пользователя в БД
    user = User(email=email, name=name)
    db.add(user)
    db.commit()
    
    # Фоновые задачи
    background_tasks.add_task(send_welcome_email, email)
    background_tasks.add_task(log_user_registration, user.id, db)
    
    # Возвращаем ответ
    return {"user_id": user.id, "status": "Welcome email will be sent soon"}

Заключение

BackgroundTasks — это легкий и удобный способ выполнения асинхронных задач в FastAPI. Используй его для простых операций, которые не требуют гарантии выполнения. Для более сложных сценариев рассмотри Celery или RQ.