Стабильная архитектура FastAPI: роутеризация, сервисы и слой схем

FastAPI быстро приводит к рабочему прототипу — но не гарантирует архитектурную устойчивость. На практике чаще всего случается одно и то же: по мере роста проекта маршруты начинают превращаться в «контроллеры всего на свете». Там смешиваются HTTP-детали (заголовки, зависимости, ошибки), бизнес-правила, доступ к данным, преобразования моделей и даже фоновая логика. В итоге усложняется тестирование, растёт стоимость изменений и становится трудно переносить доменную часть между API-процессами, задачами очередей и, например, CLI.

Эта статья — про то, как организовать FastAPI-проект так, чтобы:

• роуты оставались тонкими (HTTP-слой),
• доменная логика была вынесена в сервисы (application/domain),
• схемы (Pydantic) были отдельным слоем и служили контрактом,
• бизнес-правила тестировались без поднятия сервера и без зависимости от механики HTTP.

Подход показан на конкретных примерах кода и с учётом типичных подводных камней.

Проблема: почему маршруты превращаются в «контроллеры всего»

В «естественной» эволюции проекта разработчик добавляет endpoint за endpoint’ом. На старте всё просто: в роуте делается проверка входных данных, вызывается функция доступа к БД, формируется ответ. Но затем:

1. Появляются нюансы валидации, зависящие от доменной логики.
2. Нужна авторизация/аудит/трактовка ролей.
3. Возникают сложные запросы к данным и транзакции.
4. Под требования фронтенда начинают меняться поля ответа и форматы ошибок.
5. В разных местах начинают дублироваться одинаковые вычисления.

Роут — удобное место «всё собрать», потому что там уже есть Depends(), Request, Response, сессия/репозиторий и т.п. Но это создает слабое место: тестировать такой код неудобно. Приходится поднимать приложение, имитировать токены, проверять статус-коды и форматы ошибок — вместо того чтобы тестировать бизнес-правила как чистые функции/методы.

Результат: API «держит» домен, а не наоборот.

Целевой дизайн: слои и их ответственность

Устойчивую структуру можно описать простой матрицей ответственности.

HTTP-слой (роуты и зависимости)

Отвечает за:

• парсинг входа (Pydantic-схемы),
• привязку зависимостей (Depends),
• получение пользователя/контекста запроса,
• преобразование результатов сервисов в HTTP-ответ,
• маппинг доменных ошибок на HTTP-статусы.

Не должен отвечать за:

• бизнес-правила (вычисления, решения),
• сложные транзакционные сценарии,
• конкретные SQL/ORM-детали (это ниже по слоям).

Application/Service слой

Отвечает за:

• сценарии использования (use cases),
• доменную логику,
• работу с репозиториями/хранилищами,
• транзакционность и консистентность бизнес-данных.

Не должен знать:

• как устроен FastAPI, Request, Depends,
• как именно сериализуются ответы в JSON,
• конкретные HTTP-статусы.

Слой схем (Pydantic контракты)

Отвечает за:

• входные/выходные контракты API,
• нормализацию форматов (например, даты, идентификаторы),
• отдельное хранение схем запроса и схем ответа.

Важно: схемы не должны содержать бизнес-логики. Они — «контракты и преобразования», не «мозг системы».

Данные (репозитории/DAO)

Отвечают за:

• доступ к данным,
• выборку/сохранение,
• абстракцию над ORM/SQL,
• преобразование между ORM-моделями и доменными сущностями/DTO (если вы их используете).

Роутеризация: как проектировать модули API

В FastAPI часто используют структуру APIRouter по ресурсам: /users, /orders, /products. Но одной группировки по путям недостаточно: важно, чтобы в модуле маршрутов не росла доменная логика.

Правило тонких роутов

Endpoint должен содержать примерно такой «скелет»:

1. Получить вход (Pydantic схема).
2. Получить контекст через Depends() (например, текущий пользователь).
3. Вызвать сервис.
4. Вернуть схему ответа.

Никаких try/except с доменной логикой прямо в роуте (кроме маппинга ошибок на HTTP).

Пример структуры каталогов

app/
  main.py
  api/
    deps.py
    routers/
      users.py
      auth.py
  schemas/
    users.py
    errors.py
  services/
    users.py
  domain/
    models.py
    errors.py
  repositories/
    users.py

Это базовая схема. В реальных проектах уровни domain/application иногда объединяют, но принцип остаётся: HTTP не смешивается с доменом.

Слой схем: разделяйте вход, выход и внутренние модели

FastAPI удобно «скармливать» Pydantic-схемы напрямую. Однако для устойчивости лучше разделять:

• UserCreate — входная схема для запроса,
• UserRead — выходная схема для ответа,
• внутреннюю модель User (domain) — с минимальным набором полей под доменную логику.

Да, можно обойтись без domain-моделей и делать сервисы через схемы, но тогда доменная логика будет тесно связана с JSON-контрактами.

Пример схем для пользователя

# app/schemas/users.py
from pydantic import BaseModel, EmailStr

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

class UserRead(BaseModel):
    id: str
    email: EmailStr
    full_name: str

class UserUpdate(BaseModel):
    full_name: str

Внутреннюю доменную сущность можно описать так:

# app/domain/models.py
from dataclasses import dataclass

@dataclass(frozen=True)
class User:
    id: str
    email: str
    full_name: str

Преобразование можно делать в сервисе или в репозитории (зависит от вашей архитектуры). Главное — не держать доменную логику внутри Pydantic-схем.

Сервисный слой: use cases вместо «мешка функций»

Сервис отвечает за сценарий. Не обязательно выделять отдельный класс на каждый метод, но полезно держать единый смысловой модуль.

Пример доменных ошибок

# app/domain/errors.py
class DomainError(Exception):
    """Базовая ошибка домена."""

class EmailAlreadyUsed(DomainError):
    pass

class UserNotFound(DomainError):
    pass

Сервисы будут бросать доменные исключения. Роуты или обработчики ошибок — маппить их в HTTP.

Репозиторий как абстракция данных

# app/repositories/users.py
from typing import Protocol, Optional
from app.domain.models import User

class UsersRepo(Protocol):
    async def get_by_email(self, email: str) -> Optional[User]:
        ...

    async def get_by_id(self, user_id: str) -> Optional[User]:
        ...

    async def create(self, email: str, full_name: str) -> User:
        ...

На практике репозиторий может использовать SQLAlchemy/asyncpg и т.д. Здесь важно интерфейсное мышление: сервисы тестируются, подставляя фейковую реализацию.

Сервис: бизнес-правило — не в роуте

# app/services/users.py
from app.domain.errors import EmailAlreadyUsed, UserNotFound
from app.domain.models import User
from app.repositories.users import UsersRepo

class UsersService:
    def __init__(self, users_repo: UsersRepo):
        self._users_repo = users_repo

    async def register_user(self, *, email: str, full_name: str) -> User:
        existing = await self._users_repo.get_by_email(email)
        if existing is not None:
            raise EmailAlreadyUsed(f"Email {email} уже используется")

        # Здесь могут быть дополнительные доменные правила:
        # нормализация имени, проверка длины, политики и т.п.
        user = await self._users_repo.create(email=email, full_name=full_name)
        return user

    async def get_user(self, user_id: str) -> User:
        user = await self._users_repo.get_by_id(user_id)
        if user is None:
            raise UserNotFound(f"Пользователь {user_id} не найден")
        return user

Обратите внимание: сервис не знает, что это HTTP. Он оперирует доменными сущностями и исключениями.

Роуты: тонкая обвязка вокруг сервисов

Теперь покажем, как это выглядит в FastAPI.

Зависимости (Depends) — сборка сервисов

Один из самых удобных подходов — создать зависимости, которые строят сервис с нужной реализацией репозитория.

# app/api/deps.py
from typing import Callable
from app.repositories.users import UsersRepo
from app.services.users import UsersService

def get_users_repo() -> UsersRepo:
    # Здесь обычно создаётся репозиторий с session/engine.
    # Для примера оставим заглушку.
    raise NotImplementedError

def get_users_service() -> UsersService:
    return UsersService(users_repo=get_users_repo())

Если у вас async ORM, get_users_repo и get_users_service тоже могут быть async def, возвращающим объекты.

Роутер модуля пользователей

# app/api/routers/users.py
from fastapi import APIRouter, Depends, status
from app.schemas.users import UserCreate, UserRead
from app.services.users import UsersService
from app.domain.errors import EmailAlreadyUsed, UserNotFound
from pydantic import BaseModel

router = APIRouter(prefix="/users", tags=["users"])

# Можно использовать отдельную схему ошибок, но для демонстрации достаточно BaseModel
class ErrorResponse(BaseModel):
    detail: str

@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def register_user(
    payload: UserCreate,
    service: UsersService = Depends(  # noqa: B008
        # импортируйте get_users_service и используйте его
        lambda: None
    ),
):
    # Чтобы пример был читаемым, считаем, что service корректно подставлен
    try:
        user = await service.register_user(email=payload.email, full_name=payload.full_name)
    except EmailAlreadyUsed as e:
        # Маппинг доменной ошибки в HTTP
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
    return UserRead(id=user.id, email=user.email, full_name=user.full_name)

@router.get("/{user_id}", response_model=UserRead)
async def get_user(
    user_id: str,
    service: UsersService = Depends(lambda: None),
):
    try:
        user = await service.get_user(user_id=user_id)
    except UserNotFound as e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
    return UserRead(id=user.id, email=user.email, full_name=user.full_name)

Здесь показана идея маппинга ошибок. В реальном проекте лучше вынести обработчик исключений в единый exception handler (см. ниже), чтобы роуты не разрастались try/except.

Единый маппинг ошибок: уберите try/except из каждого роутера

Стабильная архитектура держит «перевод» ошибок в одном месте. Для FastAPI это делается через обработчики exception_handler.

Пример обработчика доменных ошибок

# app/api/handlers.py
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi import status
from app.domain.errors import DomainError, EmailAlreadyUsed, UserNotFound

async def domain_error_handler(request: Request, exc: DomainError):
    if isinstance(exc, EmailAlreadyUsed):
        return JSONResponse(
            status_code=status.HTTP_409_CONFLICT,
            content={"detail": str(exc)},
        )
    if isinstance(exc, UserNotFound):
        return JSONResponse(
            status_code=status.HTTP_404_NOT_FOUND,
            content={"detail": str(exc)},
        )

    # По умолчанию
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={"detail": str(exc)},
    )

Подключение в main.py:

# app/main.py
from fastapi import FastAPI
from app.api.routers.users import router as users_router
from app.domain.errors import DomainError
from app.api.handlers import domain_error_handler

app = FastAPI()

app.include_router(users_router)

app.add_exception_handler(DomainError, domain_error_handler)

Теперь роуты могут быть максимально простыми: сервис бросает доменную ошибку — единый обработчик маппит её в HTTP-ответ.

Правильная «сборка» приложения: основная точка включения роутов

main.py должен оставаться оркестратором: регистрация роутеров, конфигурация обработчиков, middleware, событий приложения.

Пример main.py без доменной логики

# app/main.py
from fastapi import FastAPI
from app.api.routers.users import router as users_router
from app.api.handlers import domain_error_handler
from app.domain.errors import DomainError

app = FastAPI(title="Example API")

app.include_router(users_router)
app.add_exception_handler(DomainError, domain_error_handler)

Смысл простой: не «все в main», а «main собирает».

Тестирование: почему сервисы становятся дешёвыми в поддержке

Главная практическая выгода от такой структуры — тестируемость.

Тест сервиса без FastAPI

# tests/test_users_service.py
import pytest
from app.services.users import UsersService
from app.domain.errors import EmailAlreadyUsed, UserNotFound
from app.domain.models import User

class FakeUsersRepo:
    def __init__(self):
        self.by_email = {}
        self.by_id = {}

    async def get_by_email(self, email: str):
        return self.by_email.get(email)

    async def get_by_id(self, user_id: str):
        return self.by_id.get(user_id)

    async def create(self, email: str, full_name: str):
        user = User(id="u1", email=email, full_name=full_name)
        self.by_email[email] = user
        self.by_id[user.id] = user
        return user

@pytest.mark.asyncio
async def test_register_user_success():
    repo = FakeUsersRepo()
    service = UsersService(users_repo=repo)

    user = await service.register_user(email="a@test.com", full_name="Alice")
    assert user.email == "a@test.com"
    assert user.full_name == "Alice"

@pytest.mark.asyncio
async def test_register_user_email_already_used():
    repo = FakeUsersRepo()
    repo.by_email["a@test.com"] = User(id="u1", email="a@test.com", full_name="Alice")
    service = UsersService(users_repo=repo)

    with pytest.raises(EmailAlreadyUsed):
        await service.register_user(email="a@test.com", full_name="Alice2")

@pytest.mark.asyncio
async def test_get_user_not_found():
    repo = FakeUsersRepo()
    service = UsersService(users_repo=repo)

    with pytest.raises(UserNotFound):
        await service.get_user("missing")

Такой тест не поднимает приложение, не требует TestClient, не зависит от HTTP. Он проверяет бизнес-правило напрямую.

Отдельно тестируйте контракт API

Маршруты всё равно стоит тестировать, но уже поверх сервисов: статус-коды, формат ответов, маппинг схем. Это снижает количество «хрупких» тестов, завязанных на детали домена.

Подводные камни: что ломает архитектуру на практике

1) Сервис «знает про JSON»

Проверьте, чтобы сервисы не возвращали dict ради удобства и не зависели от схем ответа. Верните доменную сущность или внутренний DTO, а контрактный слой пусть преобразует в Pydantic-схемы.

2) Роуты маппят доменные исключения вручную в каждом endpoint

Это приводит к дублированию. Лучше один exception handler на уровне приложения.

3) Pydantic-схемы превращаются в «мини-агенты»

Если внутри схем появляются методы, сложные вычисления, транзакционные решения — начнут смешиваться обязанности. Схемы должны быть декларативными: валидация, структура, преобразования типов.

4) Репозиторий возвращает ORM-модели наружу

Это не всегда плохо, но повышает связанность. Если сервисы начинают «разбираться» в ORM-объектах, тесты становятся дороже. Идеально — чтобы репозитории возвращали доменные сущности или хотя бы нейтральные структуры.

5) Транзакции расползаются по роутам

Транзакционность — доменная/сценарная ответственность. Роут не должен решать, где начинается и заканчивается транзакция. Сервис — правильное место: он знает сценарий.

Когда нужен слой DTO между доменом и схемами

На практике иногда удобно иметь отдельные внутренние структуры для представления данных, которые не совпадают с доменной сущностью один-в-один.

Например, доменная сущность Order сложнее, а ответ API требует агрегаций. Тогда:

• сервис делает сценарий,
• формирует DTO (внутренний read-model),
• роут сериализует DTO в OrderRead.

Это добавляет немного кода, но снижает вероятность того, что домен начнут подстраивать под фронтенд.

Автоматизация сборки зависимостей: не перегружайте Depends

Depends() в роуте удобен, но если вы начина