Компоненты приложения для API: маршруты, сервисы, репозитории и “тонкий контроллер”

Проект на FastAPI часто начинают с того, что эндпоинты и бизнес-логика пишутся рядом: @router.get(...) → внутри сразу расчёты, доступ к базе, валидация доменной модели, маппинг. Такой подход быстро “работает”, но почти неизбежно приводит к связности: при изменении правила бизнеса приходится трогать контроллеры и эндпоинты, а тестирование становится болью.

В этой статье разберём практичную архитектуру для API-приложений на FastAPI: разделение на слои маршруты (эндпоинты) → сервисы (бизнес-логика) → репозитории (доступ к данным) и роль “тонкого контроллера” (или тонкого слоя маршрутизации). Мы пройдёмся по тому, как разложить зависимости, как уменьшить связность и как менять бизнес-логику, не переписывая эндпоинты.

Подчеркну: речь не о “идеальной архитектуре”, а о наборе решений, которые на практике улучшают поддерживаемость и тестируемость. А если вам нужно структурировать знания системно, то полезно дополнить разбором курса «API на FastAPI» — но здесь мы сфокусируемся на базовых принципах и рабочей схеме.

Почему “эндпоинт = бизнес-логика” — плохая идея

FastAPI делает очень комфортным “вертикальный срез”: вы открыли файл, добавили роут, внутри написали запрос в БД, обработали ошибки, сформировали ответ. Но как только бизнес-правила начинают усложняться (скидки, ограничения, статусы, транзакционные сценарии), вы сталкиваетесь с типичными проблемами:

1. Эндпоинт знает слишком много

   • какие сущности читать,
   • как интерпретировать их состояние,
   • как создавать/обновлять записи,
   • какие исключения прокидывать,
   • как форматировать ошибки для HTTP.

2. Бизнес-логика размазана Одно правило может встречаться в 3–5 эндпоинтах, и любое изменение требует массового редактирования.

3. Сложнее тестировать Проверять бизнес-сценарии через HTTP возможно, но дорого по времени и хуже по изоляции. В идеале бизнес-логику тестируют без запуска сервера и без HTTP-слоя.

4. Слишком тесная связность с инфраструктурой Если сервис “знает” конкретику ORM, конкретные таблицы и детали запроса, то заменить PostgreSQL/ORM/запросы будет тяжело.

Отсюда и идея: держать HTTP-слой тонким, а бизнес-слой — независимым от FastAPI.

Целевая модель слоёв

Обычно удобная схема выглядит так:

• Маршруты (routes/controllers)
  Отвечают за:

  • разбор входных параметров (path/query/body),
  • валидацию входных схем,
  • вызов сервисов,
  • преобразование доменных ошибок в HTTP-ответы,
  • подготовку ответа (DTO/схемы).

• Сервисы (services)
  Отвечают за:

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

  Сервисы не должны “знать” о Request, Response, HTTPException, а также о деталях конкретного веб-фреймворка.

• Репозитории (repositories)
  Отвечают за:

  • чтение/запись данных,
  • инкапсуляцию запросов к БД,
  • отображение между доменными объектами и моделями хранения (ORM).

  Хорошая практика: репозиторий скрывает запросы, а сервисы используют интерфейс (абстракцию).

• Dependency wiring (зависимости) FastAPI поддерживает зависимостью инъекцию через Depends. Это удобно для сборки графа зависимостей: создать session, repository, service и передать в эндпоинт/контекст.

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

Архитектурный каркас: пример структуры

Рассмотрим структуру:

app/
  main.py
  api/
    routes/
      users.py
      orders.py
    deps.py
  domain/
    models.py
    errors.py
  services/
    users_service.py
    orders_service.py
  repositories/
    users_repo.py
    orders_repo.py
  infra/
    db.py
    orm_models.py

Это не единственно правильная схема, но она показывает “направление зависимостей”:

• api зависит от services
• services зависит от интерфейсов репозиториев/доменов
• repositories зависят от инфраструктуры БД/ORM
• infra содержит конкретику (engine/session, ORM-модели)

“Тонкий контроллер”: что именно держать в маршрутах

Тонкий контроллер — это не про минимализм ради минимализма. Это про границы ответственности.

Что делать в маршрутах

• Поднять входную схему (pydantic) и получить типизированные данные.
• Вызвать сервис: service.create_user(...).
• Преобразовать результат сервиса в response_model.
• Перехватить доменные ошибки и преобразовать в HTTPException (или ваш общий маппер ошибок).

Что не делать в маршрутах

• Писать SQL/ORM запросы.
• Собирать бизнес-логику из нескольких шагов (если только это не “протаскивание” входа).
• Решать, какие правила применяются: это работа сервиса.

Доменные модели и ошибки: основа для независимости

Чтобы сервисы могли быть независимыми от HTTP, стоит ввести доменные ошибки. Например:

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


class UserAlreadyExists(DomainError):
    def __init__(self, email: str):
        super().__init__(f"User with email {email} already exists")
        self.email = email


class UserNotFound(DomainError):
    def __init__(self, user_id: str):
        super().__init__(f"User {user_id} not found")
        self.user_id = user_id

Доменные ошибки затем маппятся на HTTP в маршрутах.

Доменные модели могут быть простыми dataclass/моделями Pydantic, но главное — их назначение: выражать смысл бизнеса, а не структуру таблиц.

Репозитории: интерфейс и реализация

Сервисам выгоднее зависеть от интерфейса репозитория, а не от конкретного SQLAlchemy-запроса.

Интерфейс репозитория (минимально)

# app/repositories/users_repo.py
from abc import ABC, abstractmethod
from typing import Optional

from app.domain.models import User


class UsersRepository(ABC):
    @abstractmethod
    async def get_by_email(self, email: str) -> Optional[User]:
        raise NotImplementedError

    @abstractmethod
    async def get_by_id(self, user_id: str) -> Optional[User]:
        raise NotImplementedError

    @abstractmethod
    async def create(self, user: User) -> User:
        raise NotImplementedError

Реализация на конкретной ORM

# app/infra/db.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/app"

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

# app/repositories/users_repo_postgres.py
from typing import Optional

from sqlalchemy import select

from app.domain.models import User
from app.repositories.users_repo import UsersRepository
from app.infra.orm_models import UserORM  # ORM-модель таблицы
from sqlalchemy.ext.asyncio import AsyncSession


class PostgresUsersRepository(UsersRepository):
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_email(self, email: str) -> Optional[User]:
        result = await self.session.execute(select(UserORM).where(UserORM.email == email))
        orm = result.scalar_one_or_none()
        if orm is None:
            return None
        return User(id=str(orm.id), email=orm.email, name=orm.name)

    async def get_by_id(self, user_id: str) -> Optional[User]:
        result = await self.session.execute(select(UserORM).where(UserORM.id == int(user_id)))
        orm = result.scalar_one_or_none()
        if orm is None:
            return None
        return User(id=str(orm.id), email=orm.email, name=orm.name)

    async def create(self, user: User) -> User:
        orm = UserORM(email=user.email, name=user.name)
        self.session.add(orm)
        await self.session.flush()  # получим orm.id без commit
        return User(id=str(orm.id), email=orm.email, name=orm.name)

Заметка: сервису не важно, как именно устроены таблицы и ORM; ему важно получить User и отдать его дальше.

Сервисы: бизнес-логика без HTTP и SQL

Рассмотрим сервис создания пользователя. Он работает с доменными объектами и репозиториями.

# app/services/users_service.py
from app.domain.errors import UserAlreadyExists
from app.domain.models import User
from app.repositories.users_repo import UsersRepository


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

    async def create_user(self, *, email: str, name: str) -> User:
        existing = await self.users_repo.get_by_email(email)
        if existing is not None:
            raise UserAlreadyExists(email=email)

        user = User(id="", email=email, name=name)
        created = await self.users_repo.create(user)
        return created

Ключевой момент: сервис не знает ничего о HTTPException, APIRouter, Depends и т.п. Он может тестироваться как обычная логика.

Транзакционность: где решать “commit” и где держать сессию

Частый вопрос: кто отвечает за commit — сервис или репозиторий?

Есть два практичных подхода:

1. Commit на уровне сервиса (или use-case)

   • сервис координирует сценарий,
   • после завершения вызывает session.commit().

2. Commit на уровне middleware/Dependency

   • в контексте запроса создаётся session,
   • commit выполняется после успешного завершения endpoint.

На практике для “слоистой” архитектуры удобно держать commit в одном месте, чтобы не размывать это по репозиториям. Часто используют “unit of work” в виде dependency, но можно начать проще: держать session в dependency и фиксировать транзакцию после вызова сервиса.

Ниже покажем оба варианта, но начнём с более прямого — commit делает зависимость (context manager).

Зависимости FastAPI: как собирать граф объектов

FastAPI позволяет создать dependencies, которые будут жизненным циклом управлять AsyncSession, репозиториями и сервисами.

# app/api/deps.py
from typing import AsyncIterator

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.infra.db import AsyncSessionLocal
from app.repositories.users_repo import UsersRepository
from app.repositories.users_repo_postgres import PostgresUsersRepository
from app.services.users_service import UsersService


async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except:
            await session.rollback()
            raise


def get_users_repo(session: AsyncSession = Depends(get_session)) -> UsersRepository:
    # Важно: репозиторий создаём на основе session
    return PostgresUsersRepository(session)


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

Так мы:

• создаём session на запрос,
• репозиторий берёт эту session,
• сервис использует репозиторий,
• commit/rollback централизованы.

Да, репозиторий в этом примере создаётся синхронно, но использует async session — это нормально, потому что методы репозитория асинхронные, а конструктор — нет.

Эндпоинт: минимум HTTP-логики и маппинг ошибок

Теперь роут для создания пользователя.

# app/api/routes/users.py
from fastapi import APIRouter, Depends, HTTPException, status

from pydantic import BaseModel, EmailStr

from app.domain.errors import UserAlreadyExists
from app.domain.models import User
from app.services.users_service import UsersService
from app.api.deps import get_users_service

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


class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str


class UserResponse(BaseModel):
    id: str
    email: EmailStr
    name: str

    @classmethod
    def from_domain(cls, user: User) -> "UserResponse":
        return cls(id=user.id, email=user.email, name=user.name)


@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    payload: CreateUserRequest,
    users_service: UsersService = Depends(get_users_service),
):
    try:
        user = await users_service.create_user(email=payload.email, name=payload.name)
    except UserAlreadyExists as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"User with email {e.email} already exists",
        )

    return UserResponse.from_domain(user)

Маршрут:

• принимает DTO (CreateUserRequest),
• вызывает сервис,
• ловит доменную ошибку и переводит её в HTTP-ответ.

Вся “бизнесовая” часть — в UsersService.

Почему это действительно помогает менять бизнес-логику, не трогая роуты

Представим правило: если имя пустое или короче 3 символов — это доменное правило. Вы можете добавить проверку в сервис, не меняя роут (если валидация уже на уровне Pydantic достаточна) или поменять доменную логику и доменную ошибку.

Ещё пример: добавляется сценарий “создать пользователя, если его нет, иначе обновить имя”. Если endpoint должен остаться прежним, то вы меняете только UsersService, возможно — меняете доменную ошибку/возврат результата. Роут может остаться прежним: он по-прежнему передаёт входные данные и форматирует выход.

То же самое относится к изменениям хранилища. Хотите заменить PostgreSQL на другой движок? Меняете репозитории в infra, а сервисы и маршруты могут не затронуться.

Как снизить связность: типовые ошибки

Даже при слоистой архитектуре легко “сломать” границы.

Ошибка 1: сервис начинает ловить HTTPException

Это почти всегда плохой запах. HTTP — это представление, а не домен. Оставляйте HTTP-исключения на уровне маршрутов.

Ошибка 2: репозиторий начинает “знать бизнес”

Например, репозиторий возвращает только “активных пользователей”, потому что так “нужно бизнесу”. Это приводит к тому, что бизнес-правила “уезжают” в инфраструктуру. Репозиторий должен быть переиспользуемым: он предоставляет операции доступа к данным.

Ошибка 3: сервис возвращает ORM-модели

Если сервис возвращает UserORM, это означает:

• он зависит от конкретной ORM,
• меняется схема хранения — сервис тоже страдает,
• тестирование становится тяжелее. Лучше возвращать доменные модели.

Ошибка 4: “протаскивание” запросов в сервисе

Иногда сервис превращается в “набор ORM-запросов”. В таком случае репозитории теряют смысл. Если в сервисе уже много запросов — стоит вынести часть в репозиторий.

Вариант: сервис как “use-case” с явными входами/выходами

Один из способов сделать границы сильнее — использовать отдельные классы/функции use-case. Например:

• CreateUserUseCase
• UpdateOrderStatusUseCase

Внутри use-case обычно есть явные параметры и явный доменный результат. Маршрут лишь адаптирует HTTP к use-case.

Это упрощает:

• версионирование сценариев,
• тестирование,
• расширение логики без роста размеров класса “UsersService”.

Обработка ошибок: где и как маппить доменные исключения

Текущий подход “try/except в каждом роуте” быстро превращается в дублирование. Чтобы не писать одно и то же в каждом endpoint, можно сделать общий слой маппинга.

Например, в маршрутах использовать общий helper:

# app/api/error_mapper.py
from fastapi import HTTPException, status
from app.domain.errors import UserAlreadyExists, UserNotFound, DomainError

def map_domain_error(err: DomainError) -> HTTPException:
    if isinstance(err, UserAlreadyExists):
        return HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(err),
        )
    if isinstance(err, UserNotFound):
        return HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(err),
        )
    return HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail=str(err),
    )

И тогда в роуте:

from app.api.error_mapper import map_domain_error
from app.domain.errors import DomainError

try:
    user = await users_service.create_user(email=payload.email, name=payload.name)
except DomainError as err:
    raise map_domain_error(err