Делал ли синхронные таски в FastAPI

Да, работал с синхронными задачами в FastAPI

В своей практике использовал несколько подходов для выполнения синхронных операций в асинхронном контексте FastAPI.

Встроенные таски FastAPI

Для простых фоновых задач используется встроенный механизм:

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/send-notification/")
async def send_notification(email: str):
    # Синхронная операция в фоновом режиме
    background_tasks = BackgroundTasks()
    background_tasks.add_task(send_email, email, message="Hello")
    return JSONResponse({"message": "Email scheduled"})

def send_email(email: str, message: str):
    # Синхронный код
    import smtplib
    # отправляем письмо
    pass

Celery для сложных таск-систем

Для более серьёзных систем с очередями и retry-логикой используем Celery:

from celery import Celery
from fastapi import FastAPI

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379",
    backend="redis://localhost:6379"
)

@celery_app.task
def process_payment(order_id: int):
    # Синхронная операция
    order = db.get_order(order_id)
    result = payment_gateway.charge(order.amount)
    return result

app = FastAPI()

@app.post("/orders/{order_id}/pay")
async def create_payment(order_id: int):
    # Запускаем таск асинхронно
    task = process_payment.delay(order_id)
    return {"task_id": task.id}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    task = celery_app.AsyncResult(task_id)
    return {"status": task.status, "result": task.result}

Использование run_in_threadpool

Для синхронных операций внутри асинхронных функций используем run_in_threadpool:

from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool
import time

app = FastAPI()

@app.get("/process")
async def process_data():
    # Синхронная функция, которая может блокировать
    result = await run_in_threadpool(expensive_sync_operation)
    return result

def expensive_sync_operation():
    # Синхронный код, вычисления, работа с БД
    time.sleep(5)
    return {"result": "processed"}

Практический опыт

В реальных проектах:

• Email рассылки — BackgroundTasks для простых случаев
• Платежи, интеграции — Celery с Redis
• Обработка файлов — run_in_threadpool для работы с disk I/O
• Синхронные библиотеки (requests вместо httpx) — также в threadpool

Критично помнить, что асинхронность в FastAPI требует правильного управления блокирующими операциями. Игнорирование этого приводит к зависаниям и деградации производительности серверного приложения.