Your service has to save an order to the database, send a payment to a bank and put a file into S3. These are three different external systems, each with its own API, SDK and failure behavior. If you mix all of it in one place, you get code that's hard to test and even harder to change. Out-adapters solve exactly this problem.
What an out-adapter is
In hexagonal architecture the application core (the business logic) knows nothing about databases, HTTP clients or message brokers. Instead it works with ports — abstract interfaces such as "save the order" or "register the payment".
An out-adapter is a concrete implementation of such a port for one external system. It takes a domain call, translates it into the external system's format, makes the request and returns the result back in domain types.
core/ (business logic)
└── calls PaymentPort.register(cmd)
↓
adapters/out/sber/ (out-adapter)
└── maps cmd → Sber JSON
└── makes POST /payment/rest/register.do
└── maps response → RegisterResult
↓
returns RegisterResult to core/
That's where the adapter's responsibility ends. It doesn't decide what to do with the result — that's the job of the business logic in core/.
One package per external system
The first habit worth developing: each external system lives in its own separate package adapters/out/<system>/.
src/order_service/
adapters/out/
persistence/ # SQLAlchemy + PostgreSQL
sber/ # Sber payment gateway
ok/ # alternative payment gateway
sms/ # SMS notifications
kafka/ # event publishing
s3/ # file storage
Why this matters:
- Replacing a provider doesn't break the others. Moved from one SMS provider to another — you edit only
adapters/out/sms/. The payment and database code isn't touched. - Resilience settings are separate for each. The timeout for Sber and the timeout for S3 are different numbers with different retry logic. A shared HTTP client for all external calls is a single point of failure.
- Metrics don't get mixed.
payment_sber_errors_totalandsms_smsc_errors_totalare different counters. Figuring out exactly where something broke is far easier. - Tests are isolated. A mock for Sber is spun up only in the
adapters/out/sber/tests and doesn't interfere with the SMS adapter's tests.
To avoid accidentally breaking this isolation, you can enforce it with import-linter:
# pyproject.toml
[[tool.importlinter.contracts]]
name = "adapters-independence"
type = "independence"
modules = [
"order_service.adapters.out.sber",
"order_service.adapters.out.ok",
"order_service.adapters.out.persistence",
"order_service.adapters.out.sms",
]
If one adapter accidentally imports another, CI fails immediately with a clear error.
Port and adapter: contract and implementation
The port is a Protocol in core/. It describes only what the business logic needs, in domain terms:
# core/order/port/out/payment_port.py
from typing import Protocol, runtime_checkable
from order_service.core.order.domain import RegisterCommand, RegisterResult, PaymentId
@runtime_checkable
class PaymentPort(Protocol):
async def register(self, cmd: RegisterCommand) -> RegisterResult: ...
async def cancel(self, payment_id: PaymentId) -> None: ...
The adapter implements this contract without declaring so explicitly — Python uses structural compatibility. It's enough for the class to have the required methods with the right signatures:
# adapters/out/sber/sber_client_adapter.py
import httpx
from order_service.core.order.domain import RegisterCommand, RegisterResult, PaymentId
from .sber_mapper import SberMapper
from .exceptions import SberError
class SberClientAdapter:
def __init__(self, client: httpx.AsyncClient, mapper: SberMapper) -> None:
self._client = client
self._mapper = mapper
async def register(self, cmd: RegisterCommand) -> RegisterResult:
payload = self._mapper.to_api(cmd)
try:
response = await self._client.post("/payment/rest/register.do", json=payload)
response.raise_for_status()
except httpx.HTTPError as exc:
raise SberError(f"Sber register failed: {exc}") from exc
return self._mapper.to_domain(response.json())
async def cancel(self, payment_id: PaymentId) -> None:
try:
response = await self._client.post(
"/payment/rest/reverse.do",
json={"orderId": str(payment_id.value)},
)
response.raise_for_status()
except httpx.HTTPError as exc:
raise SberError(f"Sber cancel failed: {exc}") from exc
Since Python doesn't check Protocol conformance at compile time, the conformance is pinned down with a test:
# tests/adapters/out/sber/test_sber_client_adapter.py
from order_service.core.order.port.out.payment_port import PaymentPort
from order_service.adapters.out.sber.sber_client_adapter import SberClientAdapter
def test_sber_adapter_satisfies_protocol() -> None:
assert issubclass(SberClientAdapter, PaymentPort)
This works because PaymentPort is marked @runtime_checkable. Python will check that the methods exist. Full signature checking is done statically by mypy.
Binding the port to the adapter happens at the application's assembly point:
# app/container.py
def build_payment_port(client: httpx.AsyncClient) -> PaymentPort:
return SberClientAdapter(client=client, mapper=SberMapper())
The Handler in core/ sees only PaymentPort — the concrete class is unknown to it.
The mapper: translation between two worlds
External systems use their own formats: kopecks instead of rubles, numeric status codes, specific fields. None of this may leak into core/ — there should be only domain types.
That's why each adapter package has a separate mapper module:
# adapters/out/sber/sber_mapper.py
from decimal import Decimal
from order_service.core.order.domain import (
RegisterCommand, RegisterResult, PaymentId, PaymentStatus, Money,
)
from order_service.adapters.out.sber.exceptions import SberError
_SBER_STATUS_MAP: dict[int, PaymentStatus] = {
0: PaymentStatus.REGISTERED,
1: PaymentStatus.AUTHORIZED,
2: PaymentStatus.DEPOSITED,
3: PaymentStatus.CANCELLED,
}
class SberMapper:
def to_api(self, cmd: RegisterCommand) -> dict:
return {
"orderNumber": str(cmd.order_id.value),
"amount": int(cmd.amount.amount * 100), # Sber accepts kopecks
"currency": 978, # 978 = RUB per ISO 4217
"description": cmd.description,
"returnUrl": str(cmd.return_url),
}
def to_domain(self, data: dict) -> RegisterResult:
raw_status = data.get("orderStatus", 0)
status = _SBER_STATUS_MAP.get(raw_status)
if status is None:
raise SberError(f"Unknown Sber status: {raw_status}")
return RegisterResult(
payment_id=PaymentId(value=data["orderId"]),
form_url=data["formUrl"],
status=status,
)
The mapper's key principles:
- It knows the specifics of its system. Kopecks, currency codes, numeric statuses — these are details of
sber/, they don't leak intocore/. - It's bidirectional.
to_api— for the request,to_domain— for the response. - Simple code. Plain
dicts, dataclasses, Pydantic models from the system's SDK. If an external system provides a Pydantic schema — use it inside the adapter; it doesn't reachcore/.
The same works for a database adapter:
# adapters/out/persistence/order_mapper.py
from order_service.core.order.domain import Order, OrderId, OrderStatus, Money
from .models import OrderRow # SQLAlchemy ORM model, not a domain type
class OrderMapper:
def to_row(self, order: Order) -> OrderRow:
return OrderRow(
id=str(order.id.value),
customer_id=str(order.customer_id.value),
status=order.status.value,
total_amount=order.total.amount,
total_currency=order.total.currency,
)
def to_domain(self, row: OrderRow) -> Order:
return Order(
id=OrderId(value=row.id),
customer_id=CustomerId(value=row.customer_id),
status=OrderStatus(row.status),
total=Money(amount=row.total_amount, currency=row.total_currency),
)
OrderRow is a SQLAlchemy ORM model. It never leaves the boundaries of adapters/out/persistence/.
Common mistake: business logic in the adapter
The adapter is responsible for one thing: translating the call there and back. The moment conditional logic about the result appears in it — that's a signal that something has gone wrong.
# Bad: the adapter makes decisions
async def register(self, cmd: RegisterCommand) -> RegisterResult:
if cmd.amount.amount > Decimal("100000"): # ← business rule
raise PaymentTooLargeError(cmd.amount)
data = await self._call_sber(cmd)
if data.get("orderStatus") == 4: # ← result interpretation
await self._notify_customer(cmd.order_id) # ← side effect
return self._mapper.to_domain(data)
Problems:
- The "100,000 is the limit" rule should live in
Order.create()or in the handler. If another payment gateway is added tomorrow, you'll have to duplicate this rule there too. _notify_customeris another port, and it's the handler that calls it. When an adapter calls another adapter, the dependency chain gets tangled.- The adapter's tests should verify mapping and network failures — not what happens after the system responds.
Correct:
# Good: the adapter only maps and calls
async def register(self, cmd: RegisterCommand) -> RegisterResult:
payload = self._mapper.to_api(cmd)
try:
response = await self._client.post("/payment/rest/register.do", json=payload)
response.raise_for_status()
except httpx.HTTPError as exc:
raise SberError(f"Sber register failed: {exc}") from exc
return self._mapper.to_domain(response.json())
Common mistake: an adapter inside an adapter
Sometimes you want to do this: "if Sber doesn't respond, let's try another gateway." And the temptation is to put this logic right into SberClientAdapter, passing an OkClientAdapter into it.
# Bad: the adapter knows about another adapter
class SberClientAdapter:
def __init__(self, ok_adapter: OkClientAdapter) -> None:
self._ok = ok_adapter
This breaks isolation. Now sber/ depends on ok/, and import-linter will catch it.
The solution is to move the switching logic into the handler:
# Good: coordination in core/
class RegisterPaymentHandler:
def __init__(
self,
sber_port: SberPaymentPort,
ok_port: OkPaymentPort,
) -> None:
self._sber = sber_port
self._ok = ok_port
async def handle(self, cmd: RegisterPaymentCommand) -> Payment:
try:
return await self._sber.register(cmd)
except PaymentPortError:
return await self._ok.register(cmd) # selection logic in core
The handler injects both ports. app/container.py binds each port to its own adapter. The adapters don't know about each other.
In short
- An out-adapter is a port implementation for one external system: it maps a domain call → request, makes the call, maps the response → domain result.
- Each external system is a separate package
adapters/out/<system>/. Replacing a provider doesn't affect the other adapters. import-linterwith anindependencecontract guarantees that adapters don't import one another.- The adapter implements a
Protocolfromcore/structurally (without an explicitimplements). Conformance is verified by a test viaissubclassand by mypy. - The mapper is a separate module in the adapter's package. It knows the external system's specifics; that specificity doesn't reach
core/. - Business logic in the adapter is a mistake: the rules live in
core/, the adapter only translates calls. - Coordination of two adapters (fallback, retry between systems) belongs in the handler, not inside an adapter.
What to read next
- Adapters in — the symmetric side: how a FastAPI router maps an incoming request into a use case command.
- Ports — exactly what an out-adapter implements: a
Protocol, domain types in signatures, port exceptions. - Core layer — why
core/knows neither SQLAlchemy, nor httpx, nor FastAPI. - Bootstrap / composition root — how
app/container.pybinds ports to adapters.