FastAPI: ошибки, которые мы допускали (и вы тоже)

FastAPI — отличный фреймворк. Документация хорошая, DX приятный, скорость разработки высокая. Именно поэтому в него легко прийти, написать что-то, что «работает», и не заметить, как приложение превращается в бомбу замедленного действия. Собрали семь ошибок, которые видим в проектах регулярно — включая свои собственные.

1. Sync-функции в async-приложении

Самая частая и самая незаметная ошибка. FastAPI запускается на ASGI-сервере (Uvicorn), event loop один. Синхронная функция, вызванная из async-контекста напрямую, блокирует весь event loop на время своего выполнения.

# Плохо: блокирует event loop
import time
import requests

@app.get("/data")
async def get_data():
    # time.sleep — синхронная блокировка
    time.sleep(2)
    # requests — синхронный HTTP-клиент
    response = requests.get("https://external-api.example.com/data")
    return response.json()

Пока этот эндпоинт отрабатывает, FastAPI не может обработать ни один другой запрос. При 100 RPS и 2-секундном sleep — приложение мертво.

# Хорошо: async везде, где есть I/O
import asyncio
import httpx

@app.get("/data")
async def get_data():
    # asyncio.sleep — не блокирует
    await asyncio.sleep(2)
    # httpx с async-клиентом
    async with httpx.AsyncClient() as client:
        response = await client.get("https://external-api.example.com/data")
    return response.json()


# Если sync-код неизбежен (legacy-библиотека, CPU-задача) —
# выносим в executor
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=10)

@app.get("/heavy")
async def heavy_computation():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, sync_heavy_function)
    return {"result": result}

Правило: если функция делает I/O (БД, HTTP, файлы) — она должна быть async def. Если CPU-heavy или использует синхронную библиотеку — run_in_executor.

2. Неправильный Dependency Injection: сессии вне Depends

FastAPI имеет мощную систему DI через Depends. Распространённая ошибка — создавать сессии и соединения напрямую, минуя её.

# Плохо: глобальная сессия или сессия вне Depends
from sqlalchemy.orm import Session

# Глобальный engine — ок. Глобальная сессия — нет.
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Создаём сессию вручную — нет гарантии закрытия
    db = SessionLocal()
    user = db.query(User).filter(User.id == user_id).first()
    # Если здесь выбросит исключение — сессия не закроется
    return user

# Хорошо: сессия через Depends с гарантированным закрытием
from typing import Generator
from sqlalchemy.orm import Session

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

С Depends и yield FastAPI гарантирует выполнение кода после yield (аналог finally) даже при исключении. Сессия всегда закроется. Дополнительный бонус: зависимости кэшируются в рамках одного запроса — одна сессия на весь request lifecycle.

3. @app.on_event вместо lifespan

@app.on_event("startup") и @app.on_event("shutdown") — deprecated с FastAPI 0.93. Они работают, но у них есть проблемы: нет гарантии порядка выполнения при нескольких обработчиках, сложнее тестировать, нет явной связи между startup и shutdown одного ресурса.

# Плохо: устаревший подход
@app.on_event("startup")
async def startup():
    app.state.redis = await aioredis.create_redis_pool(REDIS_URL)
    app.state.db_pool = await asyncpg.create_pool(DATABASE_URL)

@app.on_event("shutdown")
async def shutdown():
    app.state.redis.close()
    await app.state.redis.wait_closed()
    await app.state.db_pool.close()

# Хорошо: lifespan context manager
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Всё до yield — startup
    redis = await aioredis.create_redis_pool(REDIS_URL)
    db_pool = await asyncpg.create_pool(DATABASE_URL)

    app.state.redis = redis
    app.state.db_pool = db_pool

    yield  # Приложение работает

    # Всё после yield — shutdown
    redis.close()
    await redis.wait_closed()
    await db_pool.close()


app = FastAPI(lifespan=lifespan)

Преимущество lifespan: startup и shutdown одного ресурса находятся рядом — легче читать, легче тестировать через AsyncClient с явным async with.

4. Middleware-ловушки

Middleware в FastAPI — мощный инструмент, который легко использовать неправильно.

Ловушка первая: читать тело запроса в middleware. После чтения тело «потребляется» — следующие обработчики получат пустой body.

# Плохо: тело прочитано и потеряно
@app.middleware("http")
async def log_requests(request: Request, call_next):
    body = await request.body()  # Тело прочитано
    # Следующий обработчик не получит тело!
    response = await call_next(request)
    return response

# Хорошо: читаем и восстанавливаем
@app.middleware("http")
async def log_requests(request: Request, call_next):
    body = await request.body()
    # Восстанавливаем тело через переопределение receive
    async def receive():
        return {"type": "http.request", "body": body}
    request._receive = receive

    response = await call_next(request)
    return response

Ловушка вторая: тяжёлая логика в middleware без исключений. Middleware оборачивает каждый запрос, включая статику и healthcheck. Если middleware падает с необработанным исключением — падает весь запрос без FastAPI exception handlers.

# Плохо: исключение в middleware обходит exception handlers
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    token = request.headers.get("Authorization")
    user = decode_token(token)  # Может бросить исключение
    # Если decode_token упал — FastAPI не перехватит через HTTPException
    response = await call_next(request)
    return response

# Хорошо: перехватываем явно или используем Depends
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    try:
        token = request.headers.get("Authorization")
        if token:
            request.state.user = decode_token(token)
    except Exception:
        request.state.user = None
    response = await call_next(request)
    return response

# Или ещё лучше — auth через Depends, не middleware

5. response_model без exclude_unset

FastAPI сериализует response model, включая все поля со значениями по умолчанию — даже если они явно не были установлены. Это приводит к раздутым ответам и утечке дефолтных данных.

from pydantic import BaseModel
from typing import Optional

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    avatar_url: Optional[str] = None
    bio: Optional[str] = None
    is_premium: bool = False
    settings: dict = {}

# Плохо: вернёт ВСЕ поля, включая дефолты
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).get(user_id)
    # Ответ: {"id": 1, "name": "Alex", "email": "...",
    #          "avatar_url": null, "bio": null,
    #          "is_premium": false, "settings": {}}
    return user

# Хорошо: только установленные поля
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).get(user_id)
    return JSONResponse(
        content=UserResponse.from_orm(user).dict(exclude_unset=True)
    )

# Или через response_model_exclude_unset=True
@app.get(
    "/users/{user_id}",
    response_model=UserResponse,
    response_model_exclude_unset=True
)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).get(user_id)

response_model_exclude_unset=True — параметр декоратора, который решает проблему одной строкой. Особенно важен для PATCH-эндпоинтов и для API, которые версионируются: не хочется, чтобы новые поля неожиданно появились в ответе для старых клиентов.

6. Тяжёлый startup без healthcheck

Проблема возникает при деплое: приложение запущено, порт слушает, но внутренние ресурсы ещё инициализируются. Kubernetes или load balancer решают, что сервис готов, и шлют трафик — а он падает.

# Плохо: нет разделения readiness и liveness
@app.get("/health")
async def health():
    return {"status": "ok"}
# Этот эндпоинт вернёт 200 даже если БД ещё не подключена


# Хорошо: явный readiness check
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

class AppState:
    is_ready: bool = False
    db_pool = None
    redis = None

state = AppState()

@asynccontextmanager
async def lifespan(app: FastAPI):
    state.db_pool = await asyncpg.create_pool(DATABASE_URL)
    state.redis = await aioredis.create_redis_pool(REDIS_URL)
    state.is_ready = True  # Только после инициализации всех ресурсов
    yield
    await state.db_pool.close()
    state.redis.close()

# Liveness: приложение живо
@app.get("/healthz/live")
async def liveness():
    return {"status": "alive"}

# Readiness: готово принимать трафик
@app.get("/healthz/ready")
async def readiness():
    if not state.is_ready:
        return JSONResponse(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            content={"status": "not ready"}
        )
    # Опционально: проверяем сами ресурсы
    try:
        await state.db_pool.fetchval("SELECT 1")
    except Exception:
        return JSONResponse(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            content={"status": "db unavailable"}
        )
    return {"status": "ready"}

В Kubernetes: livenessProbe → /healthz/live, readinessProbe → /healthz/ready. Разделение критично при rolling deployments с долгим startup (загрузка ML-моделей, прогрев кэша).

7. Отсутствие структуры проекта

FastAPI не навязывает структуру — и это ловушка для тех, кто пришёл с Flask. Всё в одном файле main.py работает до первых 500 строк, потом становится неуправляемым.

# Плохо: всё в одном файле
main.py  # 2000 строк, все роуты, модели, бизнес-логика

# Хорошо: структура по доменам
app/
├── main.py           # Только создание FastAPI app и подключение роутеров
├── config.py         # Settings через pydantic-settings
├── lifespan.py       # Lifecycle management
├── dependencies.py   # Общие зависимости (get_db, get_current_user)
├── middleware.py      # Кастомные middleware
│
├── users/
│   ├── router.py     # APIRouter для /users
│   ├── schemas.py    # Pydantic-модели (Request/Response)
│   ├── models.py     # SQLAlchemy ORM модели
│   ├── service.py    # Бизнес-логика
│   └── dependencies.py  # Зависимости, специфичные для users
│
├── orders/
│   ├── router.py
│   ├── schemas.py
│   ├── models.py
│   └── service.py
│
└── common/
    ├── exceptions.py  # Кастомные исключения и exception handlers
    └── utils.py

# main.py — минимальный и чистый
from fastapi import FastAPI
from app.lifespan import lifespan
from app.users.router import router as users_router
from app.orders.router import router as orders_router
from app.middleware import setup_middleware
from app.common.exceptions import setup_exception_handlers

app = FastAPI(lifespan=lifespan)

setup_middleware(app)
setup_exception_handlers(app)

app.include_router(users_router, prefix="/api/v1")
app.include_router(orders_router, prefix="/api/v1")

Структура по доменам масштабируется: новый домен — новая директория с теми же файлами. Роутеры, схемы, сервисы и модели каждого домена изолированы. Тестировать проще — каждый сервис тестируется отдельно.

---

FastAPI прощает много ошибок на старте, но не прощает их в продакшене. Sync в async, сессии вне Depends, отсутствие lifespan — каждая из этих ошибок безобидна при малой нагрузке и смертельна при реальной.