Фоновые задачи (BackgroundTasks)

Привет! Иногда нужно выполнить длительную операцию, но не заставлять пользователя ждать. Например, отправить письмо после регистрации, сгенерировать отчёт или обработать загруженный файл. Для таких случаев в FastAPI есть BackgroundTasks.

В этой статье мы разберём:

Что такое BackgroundTasks
Создание фоновых задач
Отправка email в фоне
Генерация отчётов
Обработка данных
Отличия от Celery

Содержание

Что нужно знать перед началом
Основная часть
Что такое BackgroundTasks
Асинхронные фоновые задачи
Отправка email в фоне
Генерация отчётов в фоне
Обработка загруженных файлов
Несколько фоновых задач
BackgroundTasks vs Celery
Полный пример
Задачи для закрепления
Нюансы и подводные камни
BackgroundTasks не подходят для длительных задач
Ошибки в фоновых задачах
Зависимости
Блокировка сервера
Частые ошибки и как их избежать
Ошибка 1: Использование time.sleep() в асинхронной фоновой задаче
Ошибка 2: Слишком длительная задача
Ошибка 3: Отсутствие логирования ошибок
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание асинхронности

Совет: BackgroundTasks выполняются после отправки ответа клиенту. Это позволяет не задерживать пользователя.

Основная часть

Что такое BackgroundTasks

BackgroundTasks — это механизм FastAPI для выполнения фоновых задач после отправки HTTP-ответа.

Особенности:

Выполняются после отправки ответа
Могут быть асинхронными и синхронными
Используют тот же процесс, что и приложение
Не подходят для очень длительных задач (минуты+)

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # Имитация отправки email
    import time
    time.sleep(2)
    print(f"Email отправлен на {email}: {message}")

@app.post("/register")
async def register_user(email: str, background_tasks: BackgroundTasks):
    # Добавляем задачу в фон
    background_tasks.add_task(send_email, email, "Добро пожаловать!")
    return {"message": "Пользователь зарегистрирован. Письмо будет отправлено в фоне."}

Совет: BackgroundTasks должны быть объявлены как параметр в эндпоинте. FastAPI автоматически внедрит их.

Асинхронные фоновые задачи

import asyncio

async def send_email_async(email: str, message: str):
    await asyncio.sleep(2)  # Имитация отправки email
    print(f"Email отправлен на {email}: {message}")

@app.post("/register")
async def register_user(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_async, email, "Добро пожаловать!")
    return {"message": "Пользователь зарегистрирован"}

Совет: Фоновые задачи могут быть как синхронными, так и асинхронными функциями.

Отправка email в фоне

from fastapi import FastAPI, BackgroundTasks
import smtplib
from email.mime.text import MIMEText

app = FastAPI()

def send_email(to_email: str, subject: str, body: str):
    # Настройки SMTP
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    sender_email = "your_email@gmail.com"
    password = "your_password"

    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = sender_email
    msg["To"] = to_email

    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(sender_email, password)
        server.sendmail(sender_email, to_email, msg.as_string())

@app.post("/send-email")
async def send_email_endpoint(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(
        send_email,
        email,
        "Привет!",
        "Это письмо отправлено в фоне."
    )
    return {"message": "Письмо отправляется в фоне"}

Генерация отчётов в фоне

import csv
from io import StringIO
from fastapi import FastAPI, BackgroundTasks, Response

app = FastAPI()

def generate_report(filename: str):
    # Генерация CSV-отчёта
    data = [
        ["Name", "Age", "City"],
        ["Anna", "25", "Moscow"],
        ["Ivan", "30", "SPB"],
        ["Peter", "20", "Kazan"]
    ]

    output = StringIO()
    writer = csv.writer(output)
    writer.writerows(data)

    with open(filename, "w", encoding="utf-8") as file:
        file.write(output.getvalue())

    print(f"Отчёт создан: {filename}")

@app.post("/generate-report")
async def create_report(background_tasks: BackgroundTasks):
    filename = f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    background_tasks.add_task(generate_report, filename)
    return {"message": "Отчёт генерируется в фоне", "filename": filename}

Обработка загруженных файлов

from fastapi import FastAPI, BackgroundTasks, UploadFile, File

app = FastAPI()

def process_file(filename: str):
    # Имитация обработки файла
    import time
    time.sleep(3)
    print(f"Файл {filename} обработан")

@app.post("/upload")
async def upload_file(
    file: UploadFile = File(...),
    background_tasks: BackgroundTasks = None
):
    content = await file.read()
    filename = f"uploaded_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{file.filename}"
    with open(filename, "wb") as f:
        f.write(content)

    background_tasks.add_task(process_file, filename)
    return {"message": "Файл загружен и будет обработан в фоне"}

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

@app.post("/process")
async def process_data(data: dict, background_tasks: BackgroundTasks):
    # Добавляем несколько задач
    background_tasks.add_task(send_notification, data["user_id"])
    background_tasks.add_task(log_action, data["action"])
    background_tasks.add_task(update_stats, data)

    return {"message": "Все задачи добавлены"}

BackgroundTasks vs Celery

Характеристика	BackgroundTasks	Celery
Сложность	Простота	Сложная настройка
Длительность	Короткие (секунды)	Длительные (минуты/часы)
Масштабирование	Нет (один процесс)	Да (отдельные воркеры)
Отказоустойчивость	Нет (потеря при падении)	Да (сохранение задач)
Требует отдельного сервиса	Нет	Да (брокер, воркеры)

Совет: BackgroundTasks подходит для быстрых задач (1-5 секунд). Для длительных — используй Celery или ARQ.

Полный пример

from fastapi import FastAPI, BackgroundTasks
from datetime import datetime
import asyncio
import logging

app = FastAPI()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def send_email(email: str, message: str):
    asyncio.sleep(1)  # Имитация
    logger.info(f"Email sent to {email}: {message}")

def generate_report(user_id: int):
    asyncio.sleep(2)
    logger.info(f"Report generated for user {user_id}")

def process_file(filename: str):
    asyncio.sleep(3)
    logger.info(f"File {filename} processed")

@app.post("/register")
async def register_user(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "User registered. Email sent in background."}

@app.post("/report/{user_id}")
async def create_report(user_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(generate_report, user_id)
    return {"message": f"Report for user {user_id} is being generated"}

@app.post("/upload")
async def upload_file(filename: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_file, filename)
    return {"message": f"File {filename} is being processed"}

Задачи для закрепления

Задача 1. Напиши эндпоинт, который отправляет email в фоне после регистрации.

Задача 2. Создай фоновую задачу для генерации отчёта.

Задача 3. Добавь логирование в фоновую задачу.

Задача 4. Обработай загруженный файл в фоне.

Задача 5. Добавь несколько фоновых задач в одном эндпоинте.

Ответы:

Задача 1.

@app.post("/register")
async def register_user(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Registered"}

Задача 2.

@app.post("/report")
async def create_report(background_tasks: BackgroundTasks):
    background_tasks.add_task(generate_report)
    return {"message": "Report generating"}

Задача 3.

def send_email(email: str, message: str):
    logger.info(f"Sending email to {email}")
    time.sleep(1)
    logger.info(f"Email sent to {email}")

Задача 4.

@app.post("/upload")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    content = await file.read()
    with open(file.filename, "wb") as f:
        f.write(content)
    background_tasks.add_task(process_file, file.filename)
    return {"message": "File uploaded"}

Задача 5.

@app.post("/process")
async def process(background_tasks: BackgroundTasks):
    background_tasks.add_task(task1)
    background_tasks.add_task(task2)
    background_tasks.add_task(task3)
    return {"message": "All tasks added"}

Нюансы и подводные камни

BackgroundTasks не подходят для длительных задач

Если задача выполняется дольше 5-10 секунд, используй Celery или ARQ. В противном случае может возникнуть тайм-аут.

Ошибки в фоновых задачах

Ошибки в фоновых задачах не видны клиенту. Используй логирование для отслеживания.

Зависимости

BackgroundTasks не поддерживают Depends (внедрение зависимостей). Передавай все данные явно.

Блокировка сервера

Не используй time.sleep() в асинхронных фоновых задачах. Используй asyncio.sleep().

Частые ошибки и как их избежать

Ошибка 1: Использование `time.sleep()` в асинхронной фоновой задаче

Неправильно:

async def task():
    time.sleep(5)   # Блокирует сервер!

Правильно:

async def task():
    await asyncio.sleep(5)   # Не блокирует

Ошибка 2: Слишком длительная задача

Используй Celery для задач > 5 секунд.

Ошибка 3: Отсутствие логирования ошибок

Логируй все ошибки в фоновых задачах.

Шпаргалка

Что нужно	Как пишется
Импорт	`from fastapi import BackgroundTasks`
Добавить задачу	`background_tasks.add_task(func, args, *kwargs)`
Синхронная задача	`def task():`
Асинхронная задача	`async def task():`

Заключение

Сегодня мы:

Узнали, что такое BackgroundTasks
Создавали фоновые задачи
Отправляли email, генерировали отчёты
Разобрали отличия от Celery

КВИЗ

Что дальше?

Следующая статья: «Работа с базами данных: SQLAlchemy + FastAPI»