← Back to the section

When a project is just starting, all the files sit in one folder and everything is clear. When it grows, a question arises: where do you put what? You can organize files by type (models/, services/, routes/) — but then all the code of one feature is smeared across different folders. Hexagonal Architecture offers a different answer: lay things out by meaning and by the direction of dependencies.

Three zones: core, adapters, app

In Hexagonal Architecture all the code is divided into three zones with clear rules.

src/<service>/
├── core/          # business logic, knows nothing about FastAPI and databases
├── adapters/      # connection to the outside world: HTTP, DB, queues
└── app/           # assembles everything, entry point

The dependency arrow is strictly one-way:

app → adapters → core

core/ doesn't know about adapters/. adapters/ don't know about each other. app/ knows about everything and connects the parts.

Why Python requires import-linter

In Java, breaking the boundaries between layers doesn't compile: if core/build.gradle.kts didn't declare a dependency on Spring, the compiler will refuse to build code with import org.springframework. In Python there's no such protection. One accidental import sqlalchemy inside core/ will pass unnoticed through the IDE, the tests and the review.

That's exactly why a Hexagonal Python project needs import-linter. It checks the boundaries between layers during the build. Without it, the architecture exists only on paper.

The contract is declared in pyproject.toml:

[tool.importlinter]
root_package = "service"

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

The layers are listed from "top" to "bottom": app may import adapters and core; adapters — only core; core — none of this list. A violation looks like this:

ImportContractViolation: Module 'service.core.order.aggregate.order'
imports 'service.adapters.out.persistence.models' — violates layers contract.

lint-imports runs in CI as a mandatory check. Without a green lint-imports the branch doesn't merge.

The full project structure

A typical service on FastAPI and SQLAlchemy:

src/<service>/
├── core/
│   └── order/
│       ├── aggregate/        # Order, OrderLine (rich domain model)
│       ├── value_object/     # Money, CustomerId
│       ├── event/            # OrderConfirmed, OrderCancelled
│       ├── port/
│       │   └── out/          # OrderRepository, PaymentPort (Protocol)
│       └── usecase/          # ConfirmOrderCommand, ConfirmOrderHandler
├── adapters/
│   ├── in/
│   │   └── http/
│   │       ├── user/         # routers for client requests
│   │       └── admin/        # routers for internal operations
│   └── out/
│       ├── persistence/      # SQLAlchemy repositories
│       ├── sber/             # httpx client to an external API
│       └── kafka/            # Kafka producers
└── app/                      # composition root
    ├── main.py               # create_app(), lifespan
    ├── container.py          # DI wiring of ports to adapters
    └── settings.py           # pydantic-settings

The minimal set for a new service is core/, adapters/out/persistence/, adapters/in/http/user/, app/. Beyond that, packages are added as new external systems and entry points appear.

core/ — the zone without infrastructure

core/ contains the business logic: aggregates, value objects, events, ports and handlers. There's no import fastapi, no import sqlalchemy, no import pydantic here. Only the standard library and your own domain types.

This gives three practical advantages:

  • Fast tests. Aggregates and handlers are tested without spinning up FastAPI, without Testcontainers, without a real database. It's an ordinary Python method — you can call it directly.
  • Portability. The same core/ can easily be wrapped into a CLI script, a Celery task or a cloud function, without changing the domain code.
  • A clear boundary. A new developer opens pyproject.toml, sees the contract and understands the rules without a separate document.

Ports (interfaces for external systems) are declared as Protocol:

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

class OrderRepository(Protocol):
    async def get(self, order_id: str) -> Order: ...
    async def save(self, order: Order) -> None: ...

The handler takes ports through the constructor, knowing nothing about the implementations:

# core/order/usecase/confirm_order.py
class ConfirmOrderHandler:
    def __init__(
        self,
        orders: OrderRepository,
        payment: PaymentPort,
    ) -> None:
        self._orders = orders
        self._payment = payment

    async def handle(self, cmd: ConfirmOrderCommand) -> None:
        order = await self._orders.get(cmd.order_id)
        order.confirm()
        await self._payment.charge(order.id, order.total)
        await self._orders.save(order)

adapters/out/ — one package per external system

Each external system gets its own package in adapters/out/:

adapters/out/
├── persistence/      # PostgreSQL via SQLAlchemy (async)
├── sber/             # external payment API: httpx + DTO + mapper
├── kafka/            # aiokafka producers
└── s3/               # object storage

Each package implements one or more protocols from core/. The packages don't depend on one another. If you need to call the payment adapter after writing to the database, it's the handler in core/ that does it — it receives both ports and coordinates them.

A common mistake is to gather everything into a single file adapters/out/mega_adapter.py. Then any change in one system affects the code of the others, and the separate reliability settings (timeouts, retries) have to be mixed together.

adapters/in/ — one package per input type

Inbound adapters are also separated:

adapters/in/
├── http/
│   ├── user/         # APIRouter for client requests
│   └── admin/        # APIRouter for internal operations
└── kafka/            # consumer handlers

Why separate user/ and admin/:

  • Different access control. The user/ router checks the client's JWT; the admin/ router checks a token with a separate audience or mTLS. They can't be mixed up, because they're mounted separately with different dependencies.
  • Different request schemas. The client's CreateOrderRequest and AdminCreateOrderRequest are different Pydantic models. This rules out internal fields accidentally leaking into the client contract.
# app/main.py
from service.adapters.in_.http.user import router as user_router
from service.adapters.in_.http.admin import router as admin_router

def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    app.include_router(user_router, prefix="/api/v1", dependencies=[Depends(verify_user_jwt)])
    app.include_router(admin_router, prefix="/internal", dependencies=[Depends(verify_admin_token)])
    return app

app/ — the composition root

app/ is the assembly point. It knows about all the adapters and core/, connects them together and starts the application. There's no business logic here.

container.py binds ports to implementations:

# app/container.py
from service.adapters.out.persistence.order_repository import SqlOrderRepository
from service.adapters.out.sber.payment_adapter import SberPaymentAdapter
from service.core.order.usecase.confirm_order import ConfirmOrderHandler

class Container:
    def __init__(self, settings: Settings, session_factory) -> None:
        self.order_repo = SqlOrderRepository(session_factory)
        self.payment = SberPaymentAdapter(
            base_url=settings.sber_url,
            api_key=settings.sber_api_key,
        )
        self.confirm_order = ConfirmOrderHandler(
            orders=self.order_repo,
            payment=self.payment,
        )

lifespan manages resources — opens connections at startup and closes them at shutdown:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))   # check at startup
    yield
    await engine.dispose()

Common mistakes

No import-linter contract. The core/, adapters/ folders exist, but no one checks that the boundaries are respected. In a month, import sqlalchemy will appear in core/. The fix: add a layered contract to pyproject.toml and lint-imports to CI.

An ORM model as a domain type. The SQLAlchemy model ends up in core/ — and the architecture falls apart. The fix: in core/ there's a pure Python aggregate class; the ORM ↔ domain mapping is done in adapters/out/persistence/<x>_mapper.py.

Wiring in the adapter. adapters/out/sber/__init__.py creates the client itself and registers dependencies. The fix: adapters are pure Python classes with no side effects on import; all wiring is only in app/container.py.

User and admin routers in one package. Then you can't apply different token-checking rules without tangled conditions. The fix: create adapters/in/http/user/ and adapters/in/http/admin/ with separate dependencies right away.

In short

  • Three zones: core/ (business logic), adapters/ (connection to the outside world), app/ (assembly). The dependency arrow: app → adapters → core.
  • core/ doesn't import FastAPI, SQLAlchemy, Pydantic — only the standard library and its own types.
  • Python has no compile-time boundary protection. import-linter with a layered contract is the only tool that keeps the architecture honest. It runs in CI as a mandatory check.
  • For each external system — a separate package in adapters/out/: persistence/, sber/, kafka/. The packages don't depend on one another.
  • Different entry points — different packages in adapters/in/: http/user/ and http/admin/ are mounted with different access rules.
  • app/container.py is the only place where ports are bound to implementations. No wiring in the adapters.