Adapters In в Hexagonal на Python — FastAPI-роутеры, маппер command, Dispatcher

Опирается на правила: R-HEX-AIN-1 … R-HEX-AIN-4 и R-HEX-AIN-X1 … R-HEX-AIN-X4 из Hexagonal Style Guide → раздел 5. Adapters in.

Важно знать

• На каждый тип входа — отдельный пакет: adapters/in/http/, adapters/in/kafka/, adapters/in/cli/. Не один общий adapters/in/.
• Роутер маппит Pydantic request-DTO → UseCase command и зовёт Dispatcher. Больше ничего.
• Маппинг — отдельный файл <x>_request_mapper.py в том же пакете adapters/in/http/. Не в роутере, не в core/.
• In-adapter знает FastAPI и Pydantic. Не знает про adapters/out/* — ни persistence, ни sber, ни kafka-producer.
• Бизнес-логика в роутере запрещена (R-HEX-AIN-X1). Инварианты живут в агрегате (order.confirm()), правила — в Handler.
• Роутер зовёт Dispatcher, не репозиторий напрямую (R-HEX-AIN-X2). Иначе транзакция и авторизация ломаются.
• Роутер возвращает Pydantic response-DTO, не domain-агрегат (R-HEX-AIN-X3). Domain-тип не знает про JSON-сериализацию.
• Граница adapters/in/http/ от adapters/out/* охраняет import-linter — поставь контракт independence на in/out.

In-adapter — точка входа внешнего мира в сервис. HTTP-запрос, Kafka-сообщение или CLI-команда попадают сюда, преобразуются в UseCase command и уходят в core/ через Dispatcher. Обратно — domain-результат, который маппится в response-DTO. Раскрытие правил R-HEX-AIN-* по каждому аспекту.

Per-purpose пакеты

R-HEX-AIN-1: каждый тип входа — свой пакет.

src/orders/
  adapters/
    in/
      http/           # FastAPI-роутеры (публичный REST + admin раздельно)
        user/         # /orders/**, авторизация по JWT user-audience
        admin/        # /admin/orders/**, отдельный Depends-цепочек
      kafka/          # Consumer как entry-point (например, payment.confirmed)
      cli/            # typer-команды, batch
    out/
      persistence/
      sber/

Раздельные пакеты user/ и admin/ под http/ — это R-HEX-MOD-X3: если смешать, нельзя гарантировать, что admin-endpoint не попал в публичный роутер по ошибке. Отдельные пакеты + контракт independence в import-linter дают проверяемую изоляцию.

Полная раскладка пакетов — в Структуре модулей.

Роутер маппит request → command → response

R-HEX-AIN-2 / R-HEX-AIN-3: роутер тонкий — три строки на endpoint: маппинг → dispatch → маппинг.

# 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)


@router.get("/{order_id}", response_model=OrderResponse)
@inject
async def get_order(
    order_id: str,
    dispatcher: Dispatcher = Depends(Provide[Container.dispatcher]),
    mapper: OrderRequestMapper = Depends(Provide[Container.order_request_mapper]),
) -> OrderResponse:
    query = mapper.to_get_query(order_id)
    order = await dispatcher.dispatch(query)
    return mapper.to_response(order)

Dispatcher инжектится через dependency-injector из Container — composition root в app/. Роутер не знает, какой адаптер persistence подложен под порт репозитория.

Маппер: Pydantic DTO ↔ UseCase command

R-HEX-AIN-3: маппер — отдельный класс в adapters/in/http/. Знает Pydantic-схемы и domain-команды, больше ничего.

# adapters/in/http/user/order_request_mapper.py
from decimal import Decimal

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-схемы (CreateOrderRequest, OrderResponse) объявлены в том же пакете adapters/in/http/user/schemas.py. Они не попадают в core/ — R-HEX-CORE-X5.

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

R-HEX-AIN-4: граница знания — только web-стек.

Знает:

• fastapi — APIRouter, Depends, status, HTTPException.
• pydantic — схемы запросов и ответов.
• dependency-injector — Depends(Provide[...]) для DI.
• Dispatcher из core/ — единственная связь с доменом.
• Маппер из того же пакета.

Не знает:

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

Контракт в pyproject.toml:

[[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"]

lint-imports в CI обеспечивает выполнение этих правил автоматически — человек-ревьюер пропустит импорт, lint-imports — нет (R-HEX-TEST-X1).

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

В Python нет openapi-generator в смысле Java (где <Tag>Api генерируется из spec). Эквивалент — Pydantic-схемы, из которых 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]

Схемы описывают HTTP-контракт, живут в adapters/in/http/. Pydantic-валидация на уровне HTTP — достаточное условие для R-HEX-AIN-2: роутер не пишет if req.total_amount < 0 — это уже сделано декларативно.

Пример с Customer — Kafka-consumer как отдельный in-adapter:

# 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)

Kafka-consumer — это тоже in-adapter: принимает внешнее событие, маппит в command, зовёт Dispatcher. По той же схеме, что HTTP-роутер. Оба живут под adapters/in/, оба не знают про adapters/out/.

Что запрещено

Куда дальше

• Adapters out — симметричная сторона: как out-adapter реализует port-Protocol, маппер domain ↔ DTO внешней системы.
• Core слой — rich domain, агрегаты без FastAPI/SQLAlchemy, port как Protocol.
• Ports — как объявить outbound-порт в core/<bc>/port/out/, что попадает в сигнатуру.
• Структура модулей — полная раскладка пакетов и контракт import-linter.
• Bootstrap / composition root — как app/container.py и create_app() связывают роутеры, Dispatcher и адаптеры.
• Архитектурные тесты — lint-imports в CI как единственный надёжный guard границ.
• Когда переходить на Hexagonal — признаки готовности сервиса к Уровню 3.