← Back to the section

When an HTTP request arrives at a service, someone has to accept it, parse it and pass it into the business logic. In Hexagonal architecture this is the job of the inbound adapter (in-adapter). Let's look at how it's built in Python with FastAPI, what goes into it and what you must never do from it.

Why extract an adapter at all

Imagine a router that at the same time: validates a token, reads data from the database, calculates a discount and returns JSON. Such code is impossible to test without spinning up the whole infrastructure, and a single change breaks several things at once.

Hexagonal architecture solves this by separating responsibilities. The in-adapter is responsible for only one thing: accept an external request, translate it into the service's internal language (a command) and pass it on. The business logic, meanwhile, lives in core/ — separately, with no knowledge of FastAPI or Pydantic.

Package structure: one package per input type

The first rule is simple: each input type gets its own package. HTTP routers, Kafka consumers and CLI commands are not mixed in one folder:

src/orders/
  adapters/
    in/
      http/
        user/       # /orders/**, JWT authorization
        admin/      # /admin/orders/**, separate permission checks
      kafka/        # receiving events from other services
      cli/          # command-line commands
    out/
      persistence/
      sber/

Why separate user/ and admin/ under http/? So you don't accidentally mix public and administrative endpoints in a single router. Separate packages make such a mistake visible in review and verifiable in CI.

Thin router: three steps per endpoint

A router in Hexagonal must be as simple as possible. For each endpoint — three actions: accept the request → translate into a command → return the response.

# 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 is the router's single point of contact with the domain. The router doesn't know what happens inside: which Handler processes the command, which database is used. It just passes the command and gets a result.

The mapper: a translator between two worlds

Translating a Pydantic DTO into a command and back is a separate task, and it lives in a separate file <x>_request_mapper.py next to the router.

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

The mapper is bidirectional: to_command translates the request into an internal type, to_response translates a domain object back into Pydantic for JSON. These are different directions with different concerns: incoming data has to be validated, outgoing data has to hide what shouldn't be exposed.

Why isn't the mapper in the router? The router becomes overloaded, and the mapping logic then can't be tested separately. Why not in core/? Because Pydantic is a detail of the HTTP adapter — core/ knows nothing about it.

What the in-adapter knows and doesn't know

The inbound adapter knows only about its own technology and about how to hand control over to the domain:

Knows:

  • FastAPI — APIRouter, Depends, HTTPException
  • Pydantic — request and response schemas
  • dependency-injector — for wiring dependencies
  • Dispatcher from core/ — the single link to the domain

Doesn't know:

  • adapters/out/persistence/ — no direct work with the database
  • adapters/out/sber/ or any other outbound adapters
  • other in-packages (admin/ doesn't import from user/)

This boundary can and should be checked automatically. import-linter in CI won't let a stray import slip through unnoticed:

[[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 schemas as the API contract

In Python, Pydantic schemas play the role that in Java is taken by classes generated from OpenAPI. FastAPI automatically builds OpenAPI documentation from these schemas, and validation happens automatically — without manual checks in the router.

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

The schemas live in adapters/in/http/ — not in core/. The domain must not know about JSON serialization.

A Kafka consumer is an in-adapter too

HTTP is not the only input. A Kafka message or a CLI command works the same way: accept an external event → translate into a command → pass to the 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)

The structure is the same as for the HTTP router. This is exactly the point of Hexagonal: the domain doesn't change when a new input type is added — only the adapter changes.

Common mistakes

Business logic in the router. A check if req.total_amount > 100_000: raise HTTPException(...) is not HTTP validation, it's a business rule. It should live in the aggregate (order.create(...)) or in the Handler. In the router such a rule is invisible to domain tests and gets duplicated when a Kafka input is added.

Calling the repository directly from the router. If the router injects OrderRepository and calls repo.save(order) directly — the transaction and permission checks end up outside the Handler. That means different inputs can work with different authorization logic. Everything must go through the Dispatcher.

Returning a domain aggregate directly. The router returns Order as the response — FastAPI will try to serialize it. A domain object doesn't know about JSON, and it may have fields that must not be exposed. Always map to a Pydantic OrderResponse via a mapper.

Importing an out-adapter from an in-adapter. adapters/in/http/ must not import adapters/out/sber/. Both depend on core/, not on each other. Coordination between them happens through the Handler.

In short

  • The in-adapter is the entry point of the external world into the service: HTTP, Kafka, CLI. One input type — one package.
  • The router does three things: map the request to a command, call the Dispatcher, map the result to a response.
  • The mapper is a separate class next to the router. Not in the router, not in core/.
  • Pydantic schemas live in adapters/in/, not in core/. The domain doesn't know about JSON.
  • Business logic in the router is a mistake: it gets duplicated when a new input is added and is invisible to domain tests.
  • The router doesn't call the repository directly — only the Dispatcher.
  • The boundaries between in- and out-adapters are checked by import-linter in CI, not by a reviewer.