← назад к разделу

Когда HTTP-запрос приходит в сервис, кто-то должен его принять, разобрать и передать в бизнес-логику. В Hexagonal-архитектуре этим занимается входной адаптер (in-adapter). Разберём, как он устроен на Python с FastAPI, что в него входит и что из него категорически нельзя делать.

Зачем вообще выделять адаптер

Представьте роутер, который одновременно: проверяет токен, читает данные из базы, считает скидку и возвращает JSON. Такой код невозможно протестировать без поднятия всей инфраструктуры, и одно изменение ломает сразу несколько вещей.

Hexagonal-архитектура решает это разделением ответственностей. In-adapter отвечает только за одно: принять внешний запрос, перевести его во внутренний язык сервиса (команду) и передать дальше. Бизнес-логика при этом живёт в core/ — отдельно, без знания о FastAPI или Pydantic.

Структура пакетов: один пакет на один тип входа

Первое правило простое: у каждого типа входа — свой пакет. HTTP-роутеры, Kafka-потребители и CLI-команды не смешиваются в одной папке:

src/orders/
  adapters/
    in/
      http/
        user/       # /orders/**, авторизация по JWT
        admin/      # /admin/orders/**, отдельные проверки прав
      kafka/        # приём событий от других сервисов
      cli/          # команды командной строки
    out/
      persistence/
      sber/

Зачем разделять user/ и admin/ под http/? Чтобы случайно не смешать публичные и административные эндпоинты в одном роутере. Отдельные пакеты делают такую ошибку заметной на ревью и проверяемой в CI.

Тонкий роутер: три шага на каждый эндпоинт

Роутер в Hexagonal должен быть максимально простым. На каждый эндпоинт — три действия: принять запрос → перевести в команду → вернуть ответ.

# adapters/in/http/user/order_router.py
from fastapi import APIRouter, Depends, status
from dependency_injector.wiring import Provide, inject

from orders.app.container import Container
from orders.core.usecase.dispatcher import Dispatcher
from orders.adapters.in_.http.user.order_request_mapper import OrderRequestMapper
from orders.adapters.in_.http.user.schemas import CreateOrderRequest, OrderResponse

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


@router.post("/", status_code=status.HTTP_201_CREATED, response_model=OrderResponse)
@inject
async def create_order(
    req: CreateOrderRequest,
    dispatcher: Dispatcher = Depends(Provide[Container.dispatcher]),
    mapper: OrderRequestMapper = Depends(Provide[Container.order_request_mapper]),
) -> OrderResponse:
    command = mapper.to_command(req)
    order = await dispatcher.dispatch(command)
    return mapper.to_response(order)

Dispatcher — единственная точка связи роутера с доменом. Роутер не знает, что происходит внутри: какой Handler обрабатывает команду, какая база данных используется. Он просто передаёт команду и получает результат.

Маппер: переводчик между двумя мирами

Перевод Pydantic-DTO в команду и обратно — это отдельная задача, и она живёт в отдельном файле <x>_request_mapper.py рядом с роутером.

# adapters/in/http/user/order_request_mapper.py
from orders.core.order.command import CreateOrderCommand, GetOrderQuery
from orders.core.order.value_object import CustomerId, Money, OrderItem, Currency
from orders.adapters.in_.http.user.schemas import (
    CreateOrderRequest,
    OrderItemRequest,
    OrderResponse,
)
from orders.core.order.aggregate import Order


class OrderRequestMapper:

    def to_command(self, req: CreateOrderRequest) -> CreateOrderCommand:
        return CreateOrderCommand(
            customer_id=CustomerId(req.customer_id),
            items=[self._to_item(i) for i in req.items],
            total=Money(amount=req.total_amount, currency=Currency.RUB),
        )

    def to_get_query(self, order_id: str) -> GetOrderQuery:
        return GetOrderQuery(order_id=order_id)

    def to_response(self, order: Order) -> OrderResponse:
        return OrderResponse(
            id=str(order.id),
            status=order.status.value,
            total_amount=order.total.amount,
            customer_id=str(order.customer_id),
            items=[
                {"product_id": str(i.product_id), "quantity": i.quantity}
                for i in order.items
            ],
        )

    def _to_item(self, req: OrderItemRequest) -> OrderItem:
        return OrderItem(
            product_id=req.product_id,
            quantity=req.quantity,
            price=Money(amount=req.price, currency=Currency.RUB),
        )

Маппер двусторонний: to_command переводит запрос во внутренний тип, to_response — домен-объект обратно в Pydantic для JSON. Это разные направления с разными соображениями: входящие данные надо валидировать, исходящие — скрывать лишнее.

Почему маппер не в роутере? Роутер становится перегруженным, а логику маппинга тогда нельзя протестировать отдельно. Почему не в core/? Потому что Pydantic — это деталь HTTP-адаптера, core/ о нём не знает.

Что in-adapter знает и не знает

Входной адаптер знает только о своей технологии и о том, как передать управление домену:

Знает:

  • FastAPI — APIRouter, Depends, HTTPException
  • Pydantic — схемы запросов и ответов
  • dependency-injector — для подключения зависимостей
  • Dispatcher из core/ — единственная связь с доменом

Не знает:

  • adapters/out/persistence/ — никакой прямой работы с базой данных
  • adapters/out/sber/ или любые другие исходящие адаптеры
  • другие in-пакеты (admin/ не импортирует из user/)

Эту границу можно и нужно проверять автоматически. import-linter в CI не позволит случайному импорту проскользнуть незамеченным:

[[tool.importlinter.contracts]]
name = "in-adapters-independence"
type = "independence"
modules = [
    "orders.adapters.in_.http.user",
    "orders.adapters.in_.http.admin",
    "orders.adapters.in_.kafka",
]

[[tool.importlinter.contracts]]
name = "in-does-not-import-out"
type = "forbidden"
source_modules = ["orders.adapters.in_"]
forbidden_modules = ["orders.adapters.out"]

Pydantic-схемы как контракт API

В Python Pydantic-схемы играют роль, которую в Java берут на себя сгенерированные из OpenAPI классы. FastAPI автоматически строит OpenAPI-документацию из этих схем, а валидация происходит автоматически — без ручных проверок в роутере.

# adapters/in/http/user/schemas.py
from pydantic import BaseModel, Field
from decimal import Decimal


class OrderItemRequest(BaseModel):
    product_id: str
    quantity: int = Field(ge=1)
    price: Decimal = Field(ge=0)


class CreateOrderRequest(BaseModel):
    customer_id: str
    items: list[OrderItemRequest] = Field(min_length=1)
    total_amount: Decimal = Field(ge=0)


class OrderResponse(BaseModel):
    id: str
    status: str
    total_amount: Decimal
    customer_id: str
    items: list[dict]

Схемы живут в adapters/in/http/ — не в core/. Домен не должен знать о JSON-сериализации.

Kafka-потребитель — тоже in-adapter

HTTP — не единственный вход. Kafka-сообщение или CLI-команда работают по той же схеме: принять внешнее событие → перевести в команду → передать в Dispatcher.

# adapters/in/kafka/customer_events_consumer.py
from aiokafka import AIOKafkaConsumer
from orders.core.usecase.dispatcher import Dispatcher
from orders.adapters.in_.kafka.customer_event_mapper import CustomerEventMapper


class CustomerEventsConsumer:

    def __init__(self, dispatcher: Dispatcher, mapper: CustomerEventMapper) -> None:
        self._dispatcher = dispatcher
        self._mapper = mapper

    async def handle(self, raw: bytes) -> None:
        command = self._mapper.to_command(raw)
        await self._dispatcher.dispatch(command)

Структура та же, что у HTTP-роутера. Это и есть смысл Hexagonal: домен не меняется при добавлении нового типа входа — меняется только адаптер.

Частые ошибки

Бизнес-логика в роутере. Проверка if req.total_amount > 100_000: raise HTTPException(...) — это не валидация HTTP, это бизнес-правило. Оно должно жить в агрегате (order.create(...)) или в Handler. В роутере такое правило незаметно для тестов домена и дублируется при добавлении Kafka-входа.

Прямой вызов репозитория из роутера. Если роутер инжектит OrderRepository и вызывает repo.save(order) напрямую — транзакция и проверка прав оказываются вне Handler'а. Это значит, что разные входы могут работать с разной логикой авторизации. Всё должно идти через Dispatcher.

Возврат domain-агрегата напрямую. Роутер возвращает Order как response — FastAPI попытается его сериализовать. Domain-объект не знает о JSON, у него могут быть поля, которые нельзя показывать. Всегда маппить в Pydantic OrderResponse через маппер.

Импорт out-адаптера из in-адаптера. adapters/in/http/ не должен импортировать adapters/out/sber/. Оба зависят от core/, не друг от друга. Координация между ними — через Handler.

Коротко

  • In-adapter — точка входа внешнего мира в сервис: HTTP, Kafka, CLI. Один тип входа — один пакет.
  • Роутер делает три вещи: маппинг запроса в команду, вызов Dispatcher, маппинг результата в ответ.
  • Маппер — отдельный класс рядом с роутером. Не в роутере, не в core/.
  • Pydantic-схемы живут в adapters/in/, не в core/. Домен не знает о JSON.
  • Бизнес-логика в роутере — ошибка: она дублируется при добавлении нового входа и невидима для тестов домена.
  • Роутер не вызывает репозиторий напрямую — только Dispatcher.
  • Границы между in- и out-адаптерами проверяет import-linter в CI, не ревьюер.

Что почитать дальше