Adding an API to our Python Hexagon Architecture Template
Discussing my choice of api framework for this template and showing how our layered approach helps with testability.
One of the key benefits of a layered architecture is, you can modify layers without affecting other layers, plus you can re-use your application with what are essentially different front-end types: Desktop App, command line tool and http api - all access the same business logic and modify the same data. And when it came time to expose the application layer through HTTP, two frameworks stood out for this template: FastAPI and Litestar. Both are async-first ASGI frameworks with Pydantic support and automatic OpenAPI generation. Litestar has a more structured dependency injection system and ships more batteries — rate limiting, CSRF, response caching, a SQLAlchemy plugin that handles session lifecycle per request. FastAPI won out for one concrete reason: User is already a frozen Pydantic BaseModel, which means FastAPI serializes result.value directly with no intermediate DTO. The translation cost at the HTTP boundary is essentially zero for the user endpoints.
Route Handlers as Thin Translators
The router's only job is to translate between HTTP concepts and application layer concepts. CreateUserCommand and GetUserQuery are Mediatr GenericQuery subclasses that already exist for command-line use, so the routes just construct and dispatch them:
@router.post("/", status_code=201)
async def create_user(request: CreateUserRequest):
result = await Mediator.send_async(CreateUserCommand(request.user_id, request.name, request.email))
if result.has_errors:
raise HTTPException(status_code=400, detail=result.concat_errors())
return result.value
@router.get("/{user_id}")
async def get_user(user_id: str):
result = await Mediator.send_async(GetUserQuery(user_id))
if result.has_errors:
raise HTTPException(status_code=404, detail=result.concat_errors())
return result.value
Routing through Mediatr rather than calling UserService directly means any behaviors registered on those query types — logging, input validation, performance tracking — apply identically whether the request originates from the CLI or from an HTTP client. Adding a new cross-cutting concern means touching one behavior class, not every entry point.
Startup and Injection
The API entry point mirrors cmd/main.py exactly in how it wires the infrastructure:
@asynccontextmanager
async def lifespan(app: FastAPI):
session = await create_db_session(fresh_db=False)
def inject_layers(binder):
inject_infrastructure(binder, session=session)
inject.configure(inject_layers)
yield
app = FastAPI(title="Clean Architecture API", lifespan=lifespan)
app.include_router(users.router)
app.include_router(products.router)
The lifespan context ensures the database session exists before any request arrives and that inject is configured before the first UserService() call resolves its dependencies via @inject.autoparams().
Testing with Substitute and httpx
Since we are using Substitute3 as our mock library and a layered architecture with injected dependencies, testing our api (and our application layer) feels natural, and we do not need to think about the state of our database.
The test builds a minimal FastAPI app from the same router module, substitutes the repository through inject, and drives it with httpx.AsyncClient over ASGI — no network involved.
@pytest.fixture
def mock_repository():
mock_repo: UserRepository = Substitute()
mock_session: AsyncSession = Substitute()
def configure(binder):
binder.bind(UserRepository, mock_repo)
inject_infrastructure(binder, session=mock_session, exclude_from_binding=[UserRepository])
inject.clear()
inject.configure(configure)
yield mock_repo
inject.clear()
@pytest.mark.asyncio
async def test_returns_user_when_found(mock_repository):
user = User.from_(user_id="u1", name="Alice", email="alice@example.com")
mock_repository.get_user.returns_async(user)
async with AsyncClient(transport=ASGITransport(app=build_app()), base_url="http://test") as client:
response = await client.get("/users/u1")
assert response.status_code == 200
assert response.json()["email"] == "alice@example.com"
The fixture calls inject.clear() before configuring so it can safely replace whatever inject state was left by earlier tests. Each test configures only what it needs on the substitute — returns_async for async repository methods — and the rest of the infrastructure wiring is the real production code. The full HTTP stack from route matching through Mediatr dispatch through the service and back, including domain validation is exercised with the database as the only mock.
