You have a domain model with real behavior. You have ports that define the boundaries. You have adapters that fulfill them.
None of it runs until you wire it together.
This post covers exactly that: the application layer in hexagonal architecture, how FastAPI's dependency injection connects concrete adapters to abstract ports, and the folder structure that keeps the dependency direction honest. By the end, you'll have a fully wired hexagonal FastAPI service you can swap, test, and extend without touching your domain.
Where Does Each Piece Live?
Before writing code, fix the map. A hexagonal FastAPI project has five distinct zones:
app/
├── domain/
│ ├── models.py # Money, Discount, OrderItem, Customer, Order
│ └── ports.py # CustomerRepository, OrderRepository (Protocol)
├── adapters/
│ ├── sqlalchemy/
│ │ ├── orm.py # SQLAlchemy table definitions
│ │ └── repositories.py # SQLAlchemyCustomerRepository, SQLAlchemyOrderRepository
│ └── memory/
│ └── repositories.py # InMemoryCustomerRepository, InMemoryOrderRepository
├── application/
│ └── order_service.py # OrderService
├── api/
│ └── orders.py # FastAPI router, request/response schemas
├── dependencies.py # Dependency providers (get_session, get_order_service)
└── main.py # App factory, router registration
The rule for where something lives: if it imports from sqlalchemy, it belongs in adapters/. If it imports from fastapi, it belongs in api/. domain/ imports from nothing outside the standard library. application/ imports from domain/ only.
This layered structure is the foundation of ports and adapters each zone has a single responsibility and a clear dependency direction.
The Application Layer
OrderService is the application layer. It sits between the API and the domain: takes raw inputs, coordinates domain objects and repositories, returns a result.
No HTTP. No SQLAlchemy. It only speaks domain terms.
# app/application/order_service.py
from typing import List
from app.domain.models import Order, OrderItem
from app.domain.ports import CustomerRepository, OrderRepository
class OrderService:
def __init__(
self,
customer_repo: CustomerRepository,
order_repo: OrderRepository,
) -> None:
self.customer_repo = customer_repo
self.order_repo = order_repo
def place_order(self, customer_id: str, items: List[OrderItem]) -> Order:
customer = self.customer_repo.find_by_id(customer_id)
if not customer:
raise ValueError(f"Customer {customer_id} not found")
order = Order(
customer_id=customer_id,
items=items,
discount=customer.discount(),
).place()
return self.order_repo.save(order)
The service accepts CustomerRepository and OrderRepository Protocols from domain/ports.py. It has no idea whether those are backed by PostgreSQL, SQLite, or a Python dict. That's the point.
The application layer is what makes hexagonal architecture composable: business logic is expressed purely in domain terms, while the infrastructure details stay outside.
The SQLAlchemy Adapters
The adapters live in adapters/sqlalchemy/. They know about the database. Nothing else needs to.
# app/adapters/sqlalchemy/repositories.py
from typing import Optional
from sqlalchemy.orm import Session
from app.adapters.sqlalchemy.orm import CustomerRow, OrderRow, OrderItemRow
from app.domain.models import Customer, Money, Order, OrderItem
class SQLAlchemyCustomerRepository:
def __init__(self, session: Session) -> None:
self.session = session
def find_by_id(self, customer_id: str) -> Optional[Customer]:
row = self.session.query(CustomerRow).filter_by(id=customer_id).first()
if not row:
return None
return Customer(id=row.id, loyalty_points=row.loyalty_points)
class SQLAlchemyOrderRepository:
def __init__(self, session: Session) -> None:
self.session = session
def save(self, order: Order) -> Order:
row = OrderRow(
customer_id=order.customer_id,
status=order.status,
total=order.total().amount,
)
for item in order.items:
row.items.append(
OrderItemRow(
product_id=item.product_id,
price=item.price.amount,
quantity=item.quantity,
)
)
self.session.add(row)
self.session.flush()
order.id = str(row.id)
return order
session.flush() not session.commit() intentional. When to commit and when to roll back is the caller's job. Repositories save; they don't own transactions.
These adapters implement the ports defined in domain/ports.py. They are the concrete side of the ports and adapters pattern pluggable, replaceable, and invisible to the domain.
Wiring FastAPI with Dependency Injection
FastAPI's Depends is the dependency injection mechanism. Register a function that builds a dependency, and FastAPI calls it per request.
The wiring is split across two files to avoid a circular import: api/orders.py imports the dependency provider, and main.py imports the router they can't both import each other.
# app/dependencies.py
from typing import Generator
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from app.adapters.sqlalchemy.repositories import (
SQLAlchemyCustomerRepository,
SQLAlchemyOrderRepository,
)
from app.application.order_service import OrderService
DATABASE_URL = "postgresql://user:pass@localhost/orders_db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
def get_session() -> Generator[Session, None, None]:
session = SessionLocal()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def get_order_service(session: Session = Depends(get_session)) -> OrderService:
return OrderService(
customer_repo=SQLAlchemyCustomerRepository(session),
order_repo=SQLAlchemyOrderRepository(session),
)
# app/main.py
from fastapi import FastAPI
from app.api import orders
app = FastAPI()
app.include_router(orders.router)
get_order_service is the composition root the one place where a concrete adapter gets wired to an abstract port. This is where dependency injection makes hexagonal architecture operational. Swap SQLAlchemy for something else? dependencies.py is the only file that changes.
The FastAPI Route
The route handler is thin on purpose. It translates HTTP into domain objects, calls the service, handles errors.
# app/api/orders.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from app.application.order_service import OrderService
from app.dependencies import get_order_service
from app.domain.models import Money, OrderItem
router = APIRouter(prefix="/orders")
class OrderItemRequest(BaseModel):
product_id: str
price: float
quantity: int
class PlaceOrderRequest(BaseModel):
customer_id: str
items: List[OrderItemRequest]
class PlaceOrderResponse(BaseModel):
order_id: str
total: float
status: str
@router.post("/", response_model=PlaceOrderResponse, status_code=201)
def place_order(
body: PlaceOrderRequest,
service: OrderService = Depends(get_order_service),
) -> PlaceOrderResponse:
try:
domain_items = [
OrderItem(
product_id=i.product_id,
price=Money(i.price),
quantity=i.quantity,
)
for i in body.items
]
order = service.place_order(body.customer_id, domain_items)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc))
return PlaceOrderResponse(
order_id=order.id,
total=order.total().amount,
status=order.status,
)
Parse, call, respond. ValueError from the domain becomes a 422. Business logic errors don't reach the route the domain handles them before it gets that far.
The application layer (OrderService) absorbs all the complexity; the route stays clean because it has nothing to reason about except HTTP translation.
Swapping Adapters for Tests
Here's the same route tested without a database. The key is overriding get_order_service in FastAPI's dependency injection system.
# tests/test_orders_api.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies import get_order_service
from app.application.order_service import OrderService
from app.adapters.memory.repositories import (
InMemoryCustomerRepository,
InMemoryOrderRepository,
)
from app.domain.models import Customer
def make_test_service(customers: dict) -> OrderService:
return OrderService(
customer_repo=InMemoryCustomerRepository(customers),
order_repo=InMemoryOrderRepository(),
)
@pytest.fixture
def client_with_customer():
customers = {"c-1": Customer(id="c-1", loyalty_points=600)}
service = make_test_service(customers)
app.dependency_overrides[get_order_service] = lambda: service
yield TestClient(app)
app.dependency_overrides.clear()
def test_place_order_returns_201(client_with_customer):
response = client_with_customer.post("/orders/", json={
"customer_id": "c-1",
"items": [{"product_id": "p-1", "price": 100.0, "quantity": 2}],
})
assert response.status_code == 201
def test_place_order_applies_loyalty_discount(client_with_customer):
response = client_with_customer.post("/orders/", json={
"customer_id": "c-1",
"items": [{"product_id": "p-1", "price": 100.0, "quantity": 2}],
})
data = response.json()
assert data["total"] == 180.0
assert data["status"] == "pending"
def test_place_order_unknown_customer_returns_422():
service = make_test_service({})
app.dependency_overrides[get_order_service] = lambda: service
client = TestClient(app)
response = client.post("/orders/", json={
"customer_id": "c-999",
"items": [{"product_id": "p-1", "price": 50.0, "quantity": 1}],
})
app.dependency_overrides.clear()
assert response.status_code == 422
app.dependency_overrides swaps a dependency for the test's lifetime. No database, no env vars, no Docker. The test runs the exact same route code as production. Only the adapters differ.
The route, the service, and the domain are all tested with zero infrastructure. The SQLAlchemy* adapters get their own tests narrow, fast, and only hitting real infrastructure when you want them to.
This is dependency injection doing its job: the test harness replaces the production adapter without touching a single line of application or domain code.
Dependency Direction in Hexagonal Architecture — the Final Picture
HTTP request
↓
api/orders.py ← knows about FastAPI, Pydantic, HTTP
↓
application/order_service.py ← knows about domain only
↓
domain/ports.py (Protocol) ← the boundary; no concrete dependencies
↑
adapters/sqlalchemy/ ← knows about SQLAlchemy
adapters/memory/ ← knows about Python dicts
Nothing in domain/ imports from adapters/. Nothing in application/ imports from api/. The arrows only point inward.
This is the hexagonal architecture dependency rule enforced in code: the domain stays pure, the application layer orchestrates, and the adapters plug in at the edges.
When you reach for a Session inside OrderService, or raise HTTPException in a domain model those are the signals. A boundary is being crossed. In ports and adapters, that crossing is always a design smell.
What You've Built
Across this series, you've assembled every layer of a working hexagonal architecture in Python:
- Domain objects with real behavior no framework leakage
- Ports (Protocols) as explicit contracts at every boundary
- Adapters that implement those contracts for SQL, memory, or anything else
- An application layer that coordinates without knowing the infrastructure
-
Dependency injection via FastAPI's
Dependsto wire it all at startup - Tests that swap adapters without touching production code
The question worth asking: is this always worth it?
Post 5 is the honest answer. Trade-offs, pitfalls, when hexagonal architecture earns its complexity and when it doesn't.
Top comments (0)