The core layer in Hexagonal Architecture on Python

Python Backend

What the core layer is in Hexagonal on Python, why it doesn't depend on FastAPI and SQLAlchemy, how to declare aggregates, Protocol ports and use cases, and why import-linter is needed.

Published 28 июня 2026 ~7 min read Author: Вадим Викулин

← Back to the section

As a service grows, the business logic starts to spread out: a chunk in the FastAPI router, a chunk in the SQLAlchemy model, another chunk in the Kafka handler. It can no longer be tested without spinning up a database. Swapping the router for another framework becomes scary. This is the classic pain that Hexagonal Architecture solves.

The idea: carve out a core — a layer that contains only business logic and knows nothing about anything external. No imports from FastAPI, SQLAlchemy or any other library. Only Python.

What lives in core/

core/ is the heart of the service. Here we have:

aggregates — domain objects with business logic (Order, User, Invoice);
value objects — immutable value types (Money, CustomerId);
domain events (OrderConfirmedEvent);
exceptions (OrderNotFoundError, EmptyOrderError);
outbound ports — Protocol interfaces that describe "what the core needs from the outside world";
use cases — pairs of Command/Query + Handler.

All of this is written in pure Python. There are no imports from fastapi, sqlalchemy, pydantic, httpx, aiokafka. If such an import appears in core/ — then something is in the wrong place.

Folder structure

A typical layout looks like this:

src/orders/
  core/
    order/                           # Bounded Context "Orders"
      aggregate/
        order.py                     # Aggregate Root
      entity/
        order_item.py
      value_object/
        money.py
        customer_id.py
      event/
        order_confirmed_event.py
      exception/
        order_not_found_error.py
        empty_order_error.py
      port/
        out/
          order_repository.py        # Protocol — outbound port
          payment_port.py
    usecase/
      command/
        create_order_command.py
        confirm_order_command.py
      query/
        get_order_query.py
    service/                         # shared domain logic (rare)

The port/out/ folder contains Protocol interfaces: what the core expects from the infrastructure. They are implemented by adapters outside core/.

Rich domain versus the anemic model

A common mistake is to make the domain class just a container of fields and put all the logic into an OrderService. This is called an anemic model, and it has concrete problems:

Invariants spread out. The order confirmation logic gets duplicated in the router, in the Kafka handler, in the CLI. One of the copies will eventually fall behind.
Testing is impossible without infrastructure. An Order without methods can only be tested through a Service with a running database.
The lifecycle is unreadable. To understand when status changes, you have to search through the whole codebase rather than look at a single method.

The correct approach is a rich domain: the business logic lives inside the aggregate.

# core/order/aggregate/order.py
from __future__ import annotations
from dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING

from orders.core.order.value_object.money import Money
from orders.core.order.event.order_confirmed_event import OrderConfirmedEvent
from orders.core.order.exception.empty_order_error import EmptyOrderError
from orders.core.order.exception.invalid_status_error import InvalidStatusError

if TYPE_CHECKING:
    from orders.core.order.entity.order_item import OrderItem


class OrderStatus(StrEnum):
    DRAFT = "DRAFT"
    CONFIRMED = "CONFIRMED"
    CANCELLED = "CANCELLED"


@dataclass
class Order:
    id: str
    customer_id: str
    items: list[OrderItem]
    total: Money
    status: OrderStatus = OrderStatus.DRAFT
    _events: list = field(default_factory=list, repr=False)

    def confirm(self) -> None:
        if not self.items:
            raise EmptyOrderError(self.id)
        if self.status != OrderStatus.DRAFT:
            raise InvalidStatusError(self.status, OrderStatus.DRAFT)
        if self.total <= Money.zero():
            raise ValueError(f"Total must be positive, got {self.total}")
        self.status = OrderStatus.CONFIRMED
        self._events.append(OrderConfirmedEvent(order_id=self.id, total=self.total))

    def cancel(self, reason: str) -> None:
        if self.status == OrderStatus.CANCELLED:
            raise InvalidStatusError(self.status, "not CANCELLED")
        self.status = OrderStatus.CANCELLED

    def pop_events(self) -> list:
        events, self._events = self._events, []
        return events

The confirm() method is the one place where all the confirmation rules are described. No need to search across the whole project.

The use case — a thin orchestrator

The Handler contains no business logic. It takes the aggregate from the repository, calls a method, and saves it back:

# core/usecase/command/confirm_order_command.py
from dataclasses import dataclass
from orders.core.order.port.out.order_repository import OrderRepository


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


class ConfirmOrderCommandHandler:

    def __init__(self, order_repository: OrderRepository) -> None:
        self._order_repository = order_repository

    async def handle(self, command: ConfirmOrderCommand) -> None:
        order = await self._order_repository.find_by_id(command.order_id)
        order.confirm()                    # all logic in the aggregate
        await self._order_repository.save(order)

If an if ... else condition with business meaning appears in a Handler, that's a signal that the logic has gone to the wrong place. Move it into the aggregate.

The outbound port as a Protocol

core/ doesn't know how the repository is implemented — via SQLAlchemy, via HTTP or in-memory. It only knows the contract: "give me the Order by id" and "save the Order". The contract is declared through a Protocol:

# core/order/port/out/order_repository.py
from typing import Protocol
from orders.core.order.aggregate.order import Order


class OrderRepository(Protocol):
    async def find_by_id(self, order_id: str) -> Order: ...
    async def save(self, order: Order) -> None: ...
    async def find_all_by_customer(self, customer_id: str) -> list[Order]: ...

Protocol is structural typing. The adapter implements it without explicit inheritance: matching signatures is enough. This makes substitution in tests easier.

Domain exceptions are also declared in core/:

# core/order/exception/order_not_found_error.py
class OrderNotFoundError(Exception):
    def __init__(self, order_id: str) -> None:
        super().__init__(f"Order {order_id} not found")
        self.order_id = order_id

The Handler catches OrderNotFoundError. The FastAPI router converts it into HTTP 404. Neither core/ nor the Handler knows about HTTP — that's the adapter's job.

Dependency injection without framework annotations

Python has no Spring-style scanning. The classes in core/ are pure Python classes with __init__. No @injectable or @service decorators are needed.

Wiring happens in app/container.py outside core/:

# app/container.py
from dependency_injector import containers, providers
from orders.core.usecase.command.confirm_order_command import ConfirmOrderCommandHandler
from orders.adapters.out.persistence.order_sqlalchemy_repository import OrderSQLAlchemyRepository


class Container(containers.DeclarativeContainer):
    order_repository = providers.Singleton(OrderSQLAlchemyRepository)
    confirm_order_handler = providers.Singleton(
        ConfirmOrderCommandHandler,
        order_repository=order_repository,
    )

core/ doesn't know about Container. This lets you substitute the repository with an in-memory implementation in tests without spinning up a database:

# tests/unit/core/test_confirm_order.py
import pytest
from orders.core.order.aggregate.order import Order, OrderStatus
from orders.core.order.value_object.money import Money
from orders.core.usecase.command.confirm_order_command import (
    ConfirmOrderCommand,
    ConfirmOrderCommandHandler,
)


class InMemoryOrderRepository:
    def __init__(self, orders: dict) -> None:
        self._orders = orders

    async def find_by_id(self, order_id: str) -> Order:
        if order_id not in self._orders:
            from orders.core.order.exception.order_not_found_error import OrderNotFoundError
            raise OrderNotFoundError(order_id)
        return self._orders[order_id]

    async def save(self, order: Order) -> None:
        self._orders[order.id] = order


@pytest.mark.asyncio
async def test_confirm_order_sets_status_confirmed():
    order = Order(id="ord-1", customer_id="cust-42", items=[...], total=Money(500, "RUB"))
    repo = InMemoryOrderRepository({"ord-1": order})
    handler = ConfirmOrderCommandHandler(order_repository=repo)

    await handler.handle(ConfirmOrderCommand(order_id="ord-1"))

    assert order.status == OrderStatus.CONFIRMED

The test runs without PostgreSQL and without FastAPI — instantly.

How to protect the boundaries of core/ automatically

In Python, self-discipline doesn't scale to a team. An accidental from sqlalchemy import ... in core/ won't be noticed in review. For that there's import-linter.

# pyproject.toml
[tool.importlinter]
root_package = "orders"

[[tool.importlinter.contracts]]
name = "layers"
type = "layers"
layers = ["orders.app", "orders.adapters", "orders.core"]

[[tool.importlinter.contracts]]
name = "core-forbidden-infra"
type = "forbidden"
source_modules = ["orders.core"]
forbidden_modules = [
    "fastapi",
    "sqlalchemy",
    "pydantic",
    "httpx",
    "aiokafka",
]

Run in CI:

lint-imports

If a forbidden import appears in core/, CI will fail with a readable message even before code review. The pull request won't merge.

Common mistakes

from fastapi import HTTPException in core/. The domain layer must not know about HTTP. Declare a domain exception OrderNotFoundError in core/exception/, and do the mapping to HTTPException in the router.

A SQLAlchemy model as a domain type. If Order is a DeclarativeBase, then core/ already depends on SQLAlchemy. Keep ORM models in adapters/out/persistence/ and add a mapper between them and the domain aggregate.

A Pydantic request schema in core/. CreateOrderRequest is a REST detail. In core/ you have CreateOrderCommand (a dataclass). The Pydantic schema lives in adapters/in/http/.

A port Protocol declared in the adapter. The port is a contract from the core to the infrastructure, it belongs to core/. The adapter implements it but doesn't own it.

In short

core/ depends only on the Python standard library. No FastAPI, SQLAlchemy, Pydantic.
The business logic lives inside the aggregate (order.confirm()), not in a service class.
An outbound port is a Protocol in core/<bc>/port/out/. The adapter implements it from the outside.
The Handler is a thin orchestrator: took the aggregate, called a method, saved it.
core/ classes are pure Python classes without DI annotations. Wiring happens in app/container.py.
import-linter in CI is the only reliable way to protect the boundaries in Python.
Tests for core/ run without a database and without HTTP — fast and isolated.