Clean Architecture with FastAPI: Separating Your Domain from Infrastructure
A practical guide on how to structure FastAPI projects using clean architecture principles, with real-world examples of AWS Lambda and DynamoDB integration.
Clean Architecture with FastAPI
Python · AWS · Architecture
How to separate your domain from infrastructure in real FastAPI projects, with examples of AWS Lambda and DynamoDB integration.
A Memory from 2008
16 years ago I wrote a PHP script that read text files, formed random student groups, and rendered HTML tables — all in under 30 minutes and in a single file. It worked. But it was impossible to test, reuse, or change without breaking something. Back then there wasn’t even the vocabulary to describe it — today we’d call it a total violation of separation of concerns.
| 2008 · PHP | Today · FastAPI |
|---|---|
| Logic, data, and HTML mixed in a single block. Quick to write, impossible to maintain. | Domain, application, infrastructure, and entrypoints as independent layers. |
That mistake — mixing business logic with infrastructure details — I kept making it for years, even with modern frameworks. Routes querying the database directly, functions depending on hardcoded boto3, tests impossible to write without spinning up local DynamoDB.
The warning sign is simple: if you can’t test your business logic without touching the database, something is wrong.
Clean Architecture solves exactly this: your business rules know nothing about FastAPI, DynamoDB, or any external framework. They only know about your domain.
The Layers and What Goes Where
┌─────────────────────────────────────────────────────────────┐
│ Domain Entities · Business rules · Interfaces │
├─────────────────────────────────────────────────────────────┤
│ Application Use cases · DTOs · Domain services │
├─────────────────────────────────────────────────────────────┤
│ Infrastructure DynamoDB · S3 · Implementations │
├─────────────────────────────────────────────────────────────┤
│ Entrypoints FastAPI routes · Lambda handlers · CLI │
└─────────────────────────────────────────────────────────────┘
The core rule: dependency arrows always point inward. The domain imports nothing from the outer layers.
Project Structure
src/
domain/
entities.py # Pure Pydantic models
repositories.py # Interfaces (ABCs)
exceptions.py
application/
use_cases.py # Orchestration logic
dtos.py
infrastructure/
dynamo_repo.py # Real implementation
memory_repo.py # Implementation for tests
entrypoints/
api/ # FastAPI routers
lambda_handler.py
The Domain: Framework-Free
Entities are pure data models. The repository is an abstract interface that the domain defines but doesn’t implement.
# domain/entities.py
from dataclasses import dataclass
from datetime import datetime
@dataclass
class Order:
id: str
user_id: str
total: float
status: str
created_at: datetime
def can_be_cancelled(self) -> bool:
return self.status in ("pending", "processing")
# domain/repositories.py
from abc import ABC, abstractmethod
from .entities import Order
class OrderRepository(ABC):
@abstractmethod
async def get_by_id(self, order_id: str) -> Order | None: ...
@abstractmethod
async def save(self, order: Order) -> None: ...
Use Case: The Orchestration Logic
The use case receives the repository as a dependency. It doesn’t know if it’s DynamoDB or an in-memory dict.
# application/use_cases.py
from domain.repositories import OrderRepository
from domain.exceptions import OrderNotFound, CannotCancelOrder
class CancelOrderUseCase:
def __init__(self, repo: OrderRepository):
self.repo = repo
async def execute(self, order_id: str) -> None:
order = await self.repo.get_by_id(order_id)
if order is None:
raise OrderNotFound(order_id)
if not order.can_be_cancelled():
raise CannotCancelOrder(order.status)
order.status = "cancelled"
await self.repo.save(order)
This use case is testable with a simple
InMemoryOrderRepository. No complicated mocks or local DynamoDB needed.
Infrastructure: DynamoDB
The concrete implementation knows how to talk to DynamoDB. It implements the domain’s interface.
# infrastructure/dynamo_repo.py
import boto3
from domain.repositories import OrderRepository
from domain.entities import Order
class DynamoOrderRepository(OrderRepository):
def __init__(self, table_name: str):
self.table = boto3.resource("dynamodb").Table(table_name)
async def get_by_id(self, order_id: str) -> Order | None:
resp = self.table.get_item(Key={"id": order_id})
item = resp.get("Item")
if not item:
return None
return Order(**item)
async def save(self, order: Order) -> None:
self.table.put_item(Item=vars(order))
Entrypoint: FastAPI + Dependency Injection
FastAPI ties everything together with its Depends system. The router knows nothing about DynamoDB — it just receives the use case already configured.
# entrypoints/api/orders.py
from fastapi import APIRouter, Depends, HTTPException
from application.use_cases import CancelOrderUseCase
from infrastructure.dynamo_repo import DynamoOrderRepository
from domain.exceptions import OrderNotFound, CannotCancelOrder
import os
router = APIRouter(prefix="/orders")
def get_cancel_use_case() -> CancelOrderUseCase:
repo = DynamoOrderRepository(os.environ["ORDERS_TABLE"])
return CancelOrderUseCase(repo)
@router.delete("/{order_id}")
async def cancel_order(
order_id: str,
use_case: CancelOrderUseCase = Depends(get_cancel_use_case),
):
try:
await use_case.execute(order_id)
return {"status": "cancelled"}
except OrderNotFound:
raise HTTPException(404, "Order not found")
except CannotCancelOrder as e:
raise HTTPException(422, str(e))
The Same Handler for AWS Lambda
Here’s the real advantage: the same use case works inside a Lambda without touching a single line of business logic.
# entrypoints/lambda_handler.py
from mangum import Mangum
from fastapi import FastAPI
from entrypoints.api.orders import router
app = FastAPI()
app.include_router(router)
handler = Mangum(app) # AWS Lambda entry point
Mangumtranslates API Gateway events into the ASGI format that FastAPI understands. Without changing anything in your domain or application.
Testing: The Ultimate Reward
# tests/test_cancel_order.py
import pytest
from domain.entities import Order
from application.use_cases import CancelOrderUseCase
from datetime import datetime
class InMemoryOrderRepository:
def __init__(self, orders):
self._orders = {o.id: o for o in orders}
async def get_by_id(self, order_id):
return self._orders.get(order_id)
async def save(self, order):
self._orders[order.id] = order
@pytest.mark.asyncio
async def test_cancel_pending_order():
order = Order("1", "user-42", 99.0, "pending", datetime.now())
repo = InMemoryOrderRepository([order])
use_case = CancelOrderUseCase(repo)
await use_case.execute("1")
saved = await repo.get_by_id("1")
assert saved.status == "cancelled"
What You Gained
- Tests without a database
- Swap DynamoDB for PostgreSQL without touching business logic
- Lambda + FastAPI with the same code
- A domain readable by any developer
In Summary
Clean Architecture isn’t bureaucracy: it’s the difference between a script that works today and a system you can maintain in 5 years. The PHP from 2008 was valid for its context — quick, direct, no pretensions. But if that same code were running in production today, it would be pure technical debt.
With FastAPI the leap is especially natural because its Depends system was designed exactly for this. Start small: separate your entities into a domain/, define a repository interface, move the logic from your routes into a use case. The rest follows naturally.
Comments