Когда 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
Pydantic-схемы и есть контракт API: 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: домен не меняется при добавлении нового типа входа — меняется только адаптер.
Права на входе и кто вызывает
Проверки по признакам объекта живут в обработчике, и это только половина границы. Вторая половина остаётся адаптеру: аутентификация (подпись токена, срок, аудитория, отказ 401), грубая проверка роли на точке входа («эту ручку может вызывать только администратор», отказ 403) и защита транспорта (частота, размер запроса, взаимная проверка сертификатов). Ядро проверяет всё, что зависит от данных: «этот заказ принадлежит этому покупателю», «отменить может автор или сотрудник поддержки». Граница простая: можно ли ответить, не заглядывая в данные. Да, адаптер; нет, ядро.
Практическая заминка: правило проверяют в ядре, а контекст безопасности живёт во фреймворке, и протаскивать его в ядро нельзя. Решение это своё доменное понятие вызывающего, которое собирает адаптер и кладёт в команду:
# core/shared/caller.py: ядро знает про вызывающего на своём языке
@dataclass(frozen=True)
class Caller:
user_id: UserId
permissions: frozenset[Permission]
kind: CallerKind
def can_act_on_behalf_of(self, customer: CustomerId) -> bool: ...
# adapters/in_/http/auth.py: адаптер переводит токен в доменное понятие
def caller_from_token(claims: dict) -> Caller:
return Caller(UserId(claims["sub"]), map_permissions(claims.get("scope", "")), CallerKind.END_USER)
# adapters/in_/http/user/order_router.py: роутер кладёт вызывающего в команду
@router.post("/orders/{order_id}/cancel", status_code=204)
async def cancel_order(order_id: UUID, caller: Caller = Depends(current_caller), dispatcher: Dispatcher = Depends(get_dispatcher)) -> None:
await dispatcher.dispatch(CancelOrderCommand(OrderId(order_id), caller))
Вызывающий это часть команды, а не «контекст, который где-то есть», поэтому сценарий тестируется без инфраструктуры: передали вызывающего с нужными правами и проверили решение. Преобразование живёт в адаптере, ядро не знает ни про токены, ни про имена разрешений внешней системы. И один тип для всех входов: потребитель очереди собирает вызывающего вида «система», задача по расписанию «планировщик».
Частые ошибки
Бизнес-логика в роутере. Проверка if req.total_amount > 100_000: raise HTTPException(...) — это не валидация HTTP, это бизнес-правило. Оно должно жить в агрегате (order.create(...)) или в Handler. В роутере такое правило незаметно для тестов домена и дублируется при добавлении Kafka-входа.
# так делать не нужно
@router.post("/orders")
async def create_order(req: CreateOrderRequest, dispatcher: Dispatcher = Depends(get_dispatcher)):
if req.total_amount > 100_000: # бизнес-правило осело в роутере
raise HTTPException(status_code=400, detail="order too large")
...
Прямой вызов репозитория из роутера. Если роутер инжектит OrderRepository и вызывает repo.save(order) напрямую — транзакция и проверка прав оказываются вне Handler'а. Это значит, что разные входы могут работать с разной логикой авторизации. Всё должно идти через Dispatcher.
# так делать не нужно
@router.post("/orders")
async def create_order(req: CreateOrderRequest, orders: OrderRepository = Depends(get_order_repository)):
order = Order.create(...)
await orders.save(order) # без Handler: нет транзакции, прав и outbox
return mapper.to_response(order)
Возврат domain-агрегата напрямую. Роутер возвращает Order как response — FastAPI попытается его сериализовать. Domain-объект не знает о JSON, у него могут быть поля, которые нельзя показывать. Всегда маппить в Pydantic OrderResponse через маппер.
# так делать не нужно
@router.get("/orders/{order_id}")
async def get_order(order_id: UUID, orders: OrderRepository = Depends(get_order_repository)) -> Order:
return await orders.find_required(OrderId(order_id)) # доменная сущность в ответе
Импорт out-адаптера из in-адаптера. adapters/in_/http/ не должен импортировать adapters/out/sber/. Оба зависят от core/, не друг от друга. Координация между ними — через Handler.
Глубже: вход из очереди: чем отличается от RESTрасширенное
Заголовок обещает три вида входа, а показан в основном HTTP. У потребителя очереди своя механика и вопросы, которых у HTTP нет вовсе, и каждый решается именно в адаптере.
Нет кода ответа. Исходов у обработки три: зафиксировать смещение, пропустить как повтор, отправить в очередь недоставленных. Адаптер переводит исключения обработчика в один из них: доменная ошибка это не повтор, а DLQ с причиной; сетевая ошибка это повтор с паузой; processed_event даёт пропуск.
Смещение фиксируют вручную и после обработки, enable_auto_commit=False и await consumer.commit() после фиксации транзакции обработчика; иначе падение между ними теряет событие, о чём статья про остановку потребителя.
Порядок внутри партиции важен, между партициями нет. Если команда зависит от предыдущего события того же заказа, ключ сообщения это идентификатор заказа, и адаптер не параллелит обработку сообщений одной партиции.
Ядовитое сообщение, которое падает при каждой попытке, останавливает партицию целиком. Адаптер считает попытки по event_id и после третьей уводит сообщение в топик *.dlq с заголовками причины, а смещение фиксирует.
Маппер здесь переводит байты в команду: model_validate_json схемы события, затем в CreateOrderCommand, как to_command у роутера. Ошибка разбора это ядовитое сообщение, не повтор.
Глубже: ответу нужны данные, которых нет в агрегатерасширенное
Частая ситуация: в ответе надо показать имя покупателя, а в заказе лежит только его идентификатор, покупатели живут в другом контексте или другом сервисе. Соблазн очевидный: роутер делает второй вызов и склеивает ответ сам. Это и есть тот антипаттерн с импортом out-адаптера из in-адаптера, только в другой упаковке.
Правильные места два. Если данные нужны для чтения, это задача стороны запросов: OrderSummary из ViewRepository уже содержит customer_name, потому что витрина денормализована событиями из контекста покупателей; роутер отдаёт её как есть, об этом статья про read-model. Если данных нет в витрине и они нужны редко, за них отвечает обработчик запроса: он получает оба порта, OrderViewRepository и CustomerLookupPort, и собирает ответ в ядре, где такая координация разрешена.
# core/orders/query/get_order_card_handler.py: координация двух портов разрешена в ядре
class GetOrderCardHandler:
def __init__(self, orders: OrderViewRepository, customers: CustomerLookupPort) -> None:
self._orders = orders
self._customers = customers
async def handle(self, query: GetOrderCard) -> OrderCard:
summary = await self._orders.summary(query.order_id)
customer = await self._customers.name_of(summary.customer_id)
return OrderCard(summary=summary, customer_name=customer)
Чего не делают: не дотягивают имя покупателя в агрегат Order ради экрана и не зовут HTTP-клиент из роутера. Первое ломает границу контекстов, второе границу слоёв, и оба незаметны, пока сервис один.
Коротко
- In-adapter — точка входа внешнего мира в сервис: HTTP, Kafka, CLI. Один тип входа — один пакет.
- Роутер делает три вещи: маппинг запроса в команду, вызов
Dispatcher, маппинг результата в ответ. - Маппер — отдельный класс рядом с роутером. Не в роутере, не в
core/. - Pydantic-схемы живут в
adapters/in_/, не вcore/. Домен не знает о JSON. - Бизнес-логика в роутере — ошибка: она дублируется при добавлении нового входа и невидима для тестов домена.
- Роутер не вызывает репозиторий напрямую — только
Dispatcher. - Границы между in- и out-адаптерами проверяет
import-linterв CI, не ревьюер. - Потребитель очереди — тоже in-adapter: три исхода вместо кода ответа, ручная фиксация смещения после транзакции, ядовитые сообщения в DLQ после счётчика попыток; данные чужого контекста для ответа собирает витрина или обработчик запроса, не роутер.
Что почитать дальше
- Исходящие адаптеры (adapters out) — как out-adapter реализует port-Protocol и маппирует domain-объект во внешний DTO.
- Core-слой — domain-объекты без FastAPI и SQLAlchemy, порты как Protocol.
- Порты — как объявить outbound-порт в
core/<bc>/port/out/. - Структура модулей — полная раскладка пакетов и контракт import-linter.
- Bootstrap / composition root — как
app/container.pyсвязывает роутеры, Dispatcher и адаптеры.