Hexagonal (Ports & Adapters)

История и мотивация

Cockburn писал это в ответ на конкретную боль: приложения намертво прибиты к UI-фреймворку и БД. Тесты бизнес-логики требуют поднять браузер и БД. Переезд с Oracle на PostgreSQL — переписывание половины сервисов.

Решение: бизнес-логика ничего не знает о технологиях. Общение — только через контрактные интерфейсы (Ports), реализуемые снаружи (Adapters).

Гексагон в названии — просто фигура для рисования. Могло быть кругом, восьмиугольником, квадратом. Cockburn выбрал гексагон, потому что он даёт «естественные» стороны для разных типов адаптеров (UI, DB, тесты, …).

Ports & Adapters

Два вида портов:

• Primary (driving) port — способ войти в приложение. HTTP-контроллер, CLI-команда, message subscriber. Клиент вызывает port, port вызывает use case.
• Secondary (driven) port — способ, которым приложение обращается наружу. Repository, external API, publisher.

Идиома Python: Port через Protocol

Python 3.8+ (PEP 544) поддерживает structural subtyping через typing.Protocol. Это идеальный инструмент для Port’ов.

from typing import Protocol
from app.domain.order import Order, OrderId

class OrderRepositoryPort(Protocol):
    async def find_by_id(self, order_id: OrderId) -> Order | None: ...
    async def save(self, order: Order) -> None: ...

Ключевое: реализация не наследуется.

class PostgresOrderRepository:                          
    """Не пишем: PostgresOrderRepository(OrderRepositoryPort)."""

    def __init__(self, session: AsyncSession):
        self._session = session

    async def find_by_id(self, order_id: OrderId) -> Order | None:
        ...

    async def save(self, order: Order) -> None:
        ...

PostgresOrderRepository не декларирует зависимость от OrderRepositoryPort — но удовлетворяет его структурно. Это идеальная инверсия зависимостей: реализация не знает об интерфейсе, интерфейс не знает о реализации.

Sanity-check: type-checker (mypy, pyright) проверит совместимость на этапе связывания в composition root.

Papkovaya структура

Типичная структура Hexagonal-сервиса на Python:

app/
├── domain/              # entities, VOs, aggregates, domain services
│   ├── order.py
│   └── money.py
├── application/         # use cases + Ports
│   ├── ports/
│   │   ├── order_repository.py
│   │   ├── payment_gateway.py
│   │   └── event_publisher.py
│   └── use_cases/
│       ├── place_order.py
│       └── cancel_order.py
├── infrastructure/      # secondary adapters
│   ├── db/
│   ├── payment/
│   └── messaging/
└── interface/           # primary adapters
    ├── http/
    │   ├── routes.py
    │   └── schemas.py
    └── amqp/
        └── handlers.py

• domain/ + application/ = core (внутренность гексагона).
• infrastructure/ + interface/ = адаптеры (за границей гексагона).

Правило: domain/ и application/ не импортируют infrastructure/ и interface/. Обратное — можно.

Пример: полный поток

Разберём один сценарий сквозным путём: HTTP-запрос → use case → БД + событие.

from dataclasses import dataclass, field
from decimal import Decimal
from typing import NewType

OrderId = NewType('OrderId', int)

@dataclass
class Order:
    id: OrderId
    total: Decimal
    status: str = 'pending'

    def confirm(self) -> 'OrderConfirmed':
        if self.status != 'pending':
            raise ValueError(f'cannot confirm order in state {self.status}')
        self.status = 'confirmed'
        return OrderConfirmed(order_id=self.id)

class OrderRepositoryPort(Protocol):
    async def find_by_id(self, order_id: OrderId) -> Order | None: ...
    async def save(self, order: Order) -> None: ...

class EventPublisherPort(Protocol):
    async def publish(self, event: DomainEvent) -> None: ...

@dataclass(frozen=True)
class ConfirmOrderCommand:
    order_id: OrderId

class ConfirmOrderUseCase:
    def __init__(
        self,
        orders: OrderRepositoryPort,
        events: EventPublisherPort,
    ) -> None:
        self._orders = orders
        self._events = events

    async def execute(self, cmd: ConfirmOrderCommand) -> None:
        order = await self._orders.find_by_id(cmd.order_id)
        if order is None:
            raise OrderNotFound(cmd.order_id)
        event = order.confirm()
        await self._orders.save(order)
        await self._events.publish(event)

Что видно:

• Use case знает про Portы, ничего про SQLAlchemy или RabbitMQ.
• Order.confirm() — бизнес-инвариант живёт в domain, не в use case.
• ConfirmOrderCommand — плоский DTO, никакой связи с HTTP или AMQP.

Primary adapter:

from fastapi import APIRouter
from dishka.integrations.fastapi import FromDishka

router = APIRouter(prefix='/orders')

@router.post('/{order_id}/confirm')
async def confirm(
    order_id: int,
    use_case: FromDishka[ConfirmOrderUseCase],
) -> dict:
    await use_case.execute(ConfirmOrderCommand(order_id=OrderId(order_id)))
    return {'status': 'ok'}

Secondary adapter:

class PostgresOrderRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def find_by_id(self, order_id: OrderId) -> Order | None:
        row = await self._session.get(OrderRow, order_id)
        return _to_domain(row) if row else None

    async def save(self, order: Order) -> None:
        row = _to_orm(order)
        await self._session.merge(row)

Composition root связывает всё вместе. Порт-к-адаптеру — one-to-one, но use case видит только Port.

Тестируемость

Главная выгода Hexagonal — прозрачные тесты бизнес-логики без внешнего мира.

class InMemoryOrderRepository:                          
    def __init__(self, seed: dict[OrderId, Order] = None):
        self._store = seed or {}

    async def find_by_id(self, order_id):
        return self._store.get(order_id)

    async def save(self, order):
        self._store[order.id] = order

class CapturingPublisher:
    def __init__(self):
        self.events = []
    async def publish(self, event):
        self.events.append(event)

async def test_confirm_order_success():
    repo = InMemoryOrderRepository({OrderId(1): Order(id=OrderId(1), total=Decimal('100'))})
    publisher = CapturingPublisher()
    use_case = ConfirmOrderUseCase(orders=repo, events=publisher)

    await use_case.execute(ConfirmOrderCommand(order_id=OrderId(1)))

    assert repo._store[OrderId(1)].status == 'confirmed'
    assert len(publisher.events) == 1

Никакой SQLAlchemy, никакого RabbitMQ. Ноль моков (структурно — Protocol удовлетворяется автоматически). Тест — быстрый, детерминированный, читаемый.

Как не надо

1. Port ради Port’а

class OrderRepositoryPort(Protocol):
    async def get(self, order_id: int) -> dict: ...
    async def save(self, data: dict) -> None: ...

Порт возвращает dict. Домена нет. Port — просто «обёртка над словарём». Инверсия зависимостей — фикция, потому что и core, и adapter общаются через один и тот же неструктурированный тип.

Правильно: port возвращает Order (aggregate), а не dict. Иначе инверсия не защищает от изменений формата.

2. Adapter знает про Application

from app.application.use_cases.place_order import PlaceOrderUseCase   

class PostgresOrderRepository:
    async def save(self, order):
        # ... сохранить
        await PlaceOrderUseCase(...).notify(order)                    

Adapter вызывает use case. Стрелка зависимости перевёрнута дважды: adapter знает про core, а не наоборот. Гексагон разрушен.

Правильно: если после сохранения нужно что-то сделать — либо это делает use case, либо это порт EventPublisherPort, реализуемый другим adapter’ом.

3. Один жирный порт на все операции

class DatabasePort(Protocol):
    async def get_order(self, id): ...
    async def save_order(self, order): ...
    async def get_customer(self, id): ...
    async def save_customer(self, customer): ...
    async def get_invoice(self, id): ...
    ...

15 методов на разные сущности. Первое нарушение ISP (Interface Segregation): use case, работающий только с Order, вынужден принимать порт с методами про Customer и Invoice.

Правильно: один порт на один aggregate. OrderRepositoryPort, CustomerRepositoryPort, InvoiceRepositoryPort.

4. Domain в domain, ORM тоже в domain

from sqlalchemy.orm import Mapped, mapped_column                   
from app.infrastructure.db.base import Base

class Order(Base):                                                 
    __tablename__ = 'orders'
    id: Mapped[int] = mapped_column(primary_key=True)
    ...

Domain-класс наследуется от Base из infrastructure. Core теперь зависит от SQLAlchemy — весь смысл Hexagonal утерян. См. главу Clean Architecture — там разбор с рефакторингом.

Практика: разложите по слоям

class Order:
    def apply_discount(self, percent: Decimal) -> None:
        if percent > 50:
            raise DomainError("discount too big")

class OrderStandardizeUseCase:
    def __init__(self, normalizer: NormalizerPort, repo: ItemRepositoryPort):
        ...
    async def execute(self, msg: OrderItemMessage) -> None: ...

class ItemRepositoryPort(Protocol):
    async def find_by_id(self, item_id: int) -> Item | None: ...
    async def save(self, item: Item) -> None: ...

self._client = httpx.AsyncClient(
    timeout=60.0,
    limits=httpx.Limits(max_connections=100),
)

@broker.subscriber(queue, exchange)
async def handle_order_item(
    message: OrderItemMessage,
    use_case: FromDishka[OrderStandardizeUseCase],
) -> None:
    await use_case.execute(message)

@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: Currency

    def __add__(self, other: Money) -> Money:
        if self.currency != other.currency:
            raise DomainError()
        return Money(self.amount + other.amount, self.currency)

class PostgresItemRepository(ItemRepositoryPort):
    def __init__(self, session_factory: async_sessionmaker[AsyncSession]):
        self._session_factory = session_factory

    async def save(self, item: Item) -> None:
        async with self._session_factory() as session:
            ...

@app.post("/orders")
async def create_order(
    body: CreateOrderRequest,
    use_case: FromDishka[CreateOrderUseCase],
) -> CreateOrderResponse:
    return await use_case.execute(body.to_command())

class OutgoingMessageMapper:
    """Приводит StandardizedItem к формату matcher-service."""
    def to_matcher_format(self, item: StandardizedItem) -> dict:
        return {"raw": {...}, "values": [{...}]}

class ShippingPolicy:
    """Как считается стоимость доставки в зависимости от веса и региона."""
    def calculate(self, order: Order, region: Region) -> Money:
        ...

Trade-offs

Один нетривиальный контр-аргумент: если core — тривиальный CRUD, то Hexagonal вам даст «пустой гексагон» — вся сложность окажется в адаптерах, а внутри — просто перекладывание из формы в БД. В этом случае Layered честнее.

В твоём же коде

Разбор из анонимизированного микросервиса — практически хрестоматийная Hexagonal:

app/
├── application/
│   ├── ports/
│   │   ├── normalizer.py           # NormalizerPort
│   │   ├── repository.py           # ItemRepositoryPort
│   │   └── event_publisher.py      # EventPublisherPort
│   └── use_cases/
│       └── standardize.py          # OrderStandardizeUseCase
├── infrastructure/
│   ├── marketplace_adapter.py      # реализует NormalizerPort через httpx
│   ├── db/repository.py            # реализует ItemRepositoryPort
│   └── rabbitmq/publisher.py       # реализует EventPublisherPort
└── interface/
    └── rabbitmq/handlers.py        # primary adapter

Что тут хорошо:

• Все port’ы через Protocol.
• Реализации не декларируют зависимость от port’ов — чистая инверсия.
• Composition root на Dishka явно связывает Port → Adapter.

Что можно улучшить:

• domain/ пустой (нет aggregates). Это приемлемо для сервиса-оркестратора, но при усложнении логика начнёт скапливаться в use case.
• _adapt_for_consumer в use case — утечка контракта consumer’а в бизнес-слой. Должен быть отдельный OutgoingMessageMapper в infrastructure. Разбор — на странице Case study.

Дальнейшее чтение

• Alistair Cockburn. Hexagonal Architecture. 2005. Оригинал.
• Vaughn Vernon. Implementing Domain-Driven Design. Chapter 4. Hexagonal как основа архитектуры DDD-сервиса.
• Herberto Graça. Ports & Adapters Architecture. Разбор с картинками.
• Harry Percival, Bob Gregory. Architecture Patterns with Python. Практическая Hexagonal + DDD на Python. Бесплатно.
• Tom Hombergs. Get Your Hands Dirty on Clean Architecture. Тот же подход в Java.

Проверьте себя