Python Clean Architecture
Overview
Reference guide for structuring Python services with clean architecture. Apply these patterns to separate business logic from framework concerns, making code testable, maintainable, and framework-independent.
Project Structure
src/
├── domain/ # Core business logic — NO framework imports
│ ├── models/ # Domain entities and value objects
│ │ ├── user.py
│ │ └── order.py
│ ├── errors.py # Domain-specific exceptions
│ └── services/ # Business logic / use cases
│ ├── user_service.py
│ └── order_service.py
├── infrastructure/ # External concerns
│ ├── repositories/ # Data access implementations
│ │ ├── user_repo.py
│ │ └── order_repo.py
│ ├── external/ # Third-party API clients
│ │ └── payment_client.py
│ └── db.py # Database connection setup
├── api/ # Framework layer (FastAPI, Flask, etc.)
│ ├── routers/
│ │ ├── users.py
│ │ └── orders.py
│ ├── dependencies.py # DI wiring
│ └── error_handlers.py
└── main.pyKey rule: Dependencies point inward. domain/ imports nothing from infrastructure/ or api/. infrastructure/ imports from domain/. api/ imports from both.
Domain Models (vs ORM Models)
Domain models represent business concepts with behavior. ORM models represent database tables. Keep them separate.
# domain/models/user.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum
class UserRole(StrEnum):
MEMBER = "member"
ADMIN = "admin"
@dataclass
class User:
id: str
email: str
name: str
role: UserRole
created_at: datetime
is_active: bool = True
def promote_to_admin(self) -> None:
if not self.is_active:
raise InactiveUserError(self.id)
self.role = UserRole.ADMIN
def deactivate(self) -> None:
self.is_active = False
@property
def is_admin(self) -> bool:
return self.role == UserRole.ADMIN
# domain/models/order.py
@dataclass
class OrderItem:
product_id: str
quantity: int
unit_price: float
@property
def total(self) -> float:
return self.quantity * self.unit_price
@dataclass
class Order:
id: str
user_id: str
items: list[OrderItem] = field(default_factory=list)
status: str = "draft"
@property
def total(self) -> float:
return sum(item.total for item in self.items)
def add_item(self, product_id: str, quantity: int, unit_price: float) -> None:
if self.status != "draft":
raise OrderNotEditableError(self.id, self.status)
if quantity <= 0:
raise ValueError("Quantity must be positive")
self.items.append(OrderItem(product_id, quantity, unit_price))
def submit(self) -> None:
if not self.items:
raise EmptyOrderError(self.id)
self.status = "submitted"ORM Model — Separate Concern
ORM models mirror the database schema and may include fields not in the domain model (e.g., password_hash). Map between ORM and domain models in the repository layer.
# infrastructure/db_models/user_model.py
class UserModel(Base):
__tablename__ = "users"
id = Column(String, primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
role = Column(String, nullable=False, default="member")
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, nullable=False)
password_hash = Column(String, nullable=False) # Not in domain modelRepository Pattern
Define abstract interfaces in the domain layer. Implement in infrastructure.
# domain/repositories.py
from abc import ABC, abstractmethod
class UserRepository(ABC):
@abstractmethod
async def get_by_id(self, user_id: str) -> User | None: ...
@abstractmethod
async def get_by_email(self, email: str) -> User | None: ...
@abstractmethod
async def save(self, user: User) -> None: ...
@abstractmethod
async def delete(self, user_id: str) -> None: ...
class OrderRepository(ABC):
@abstractmethod
async def get_by_id(self, order_id: str) -> Order | None: ...
@abstractmethod
async def save(self, order: Order) -> None: ...Repository Implementation
# infrastructure/repositories/user_repo.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from domain.repositories import UserRepository
from domain.models.user import User, UserRole
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, session: AsyncSession):
self._session = session
async def get_by_id(self, user_id: str) -> User | None:
model = await self._session.get(UserModel, user_id)
return self._to_domain(model) if model else None
async def get_by_email(self, email: str) -> User | None:
result = await self._session.execute(
select(UserModel).where(UserModel.email == email)
)
model = result.scalar_one_or_none()
return self._to_domain(model) if model else None
async def save(self, user: User) -> None:
model = await self._session.get(UserModel, user.id)
if model is None:
model = UserModel(id=user.id)
self._session.add(model)
model.email = user.email
model.name = user.name
model.role = user.role.value
model.is_active = user.is_active
model.created_at = user.created_at
async def delete(self, user_id: str) -> None:
model = await self._session.get(UserModel, user_id)
if model:
await self._session.delete(model)
@staticmethod
def _to_domain(model: UserModel) -> User:
return User(
id=model.id, email=model.email, name=model.name,
role=UserRole(model.role), is_active=model.is_active,
created_at=model.created_at,
)Service Layer
Services contain business logic. They depend on repository abstractions, never on concrete implementations or frameworks.
# domain/services/user_service.py
from domain.models.user import User, UserRole
from domain.repositories import UserRepository
from domain.errors import NotFoundError, ConflictError
class UserService:
def __init__(self, user_repo: UserRepository):
self._user_repo = user_repo
async def get_user(self, user_id: str) -> User:
user = await self._user_repo.get_by_id(user_id)
if user is None:
raise NotFoundError("User", user_id)
return user
async def create_user(self, email: str, name: str) -> User:
existing = await self._user_repo.get_by_email(email)
if existing is not None:
raise ConflictError(f"User with email {email} already exists")
user = User(
id=generate_id(),
email=email,
name=name,
role=UserRole.MEMBER,
created_at=utcnow(),
)
await self._user_repo.save(user)
return user
async def promote_to_admin(self, user_id: str) -> User:
user = await self.get_user(user_id)
user.promote_to_admin() # Domain logic on the model
await self._user_repo.save(user)
return user
async def deactivate_user(self, user_id: str) -> User:
user = await self.get_user(user_id)
user.deactivate()
await self._user_repo.save(user)
return userDependency Injection (Without Framework)
Wire dependencies manually using constructor injection. No DI container needed for most Python services.
# api/dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from infrastructure.db import get_session
from infrastructure.repositories.user_repo import SqlAlchemyUserRepository
from domain.services.user_service import UserService
async def get_user_repo(
session: AsyncSession = Depends(get_session),
) -> SqlAlchemyUserRepository:
return SqlAlchemyUserRepository(session)
async def get_user_service(
user_repo: SqlAlchemyUserRepository = Depends(get_user_repo),
) -> UserService:
return UserService(user_repo)# api/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from domain.services.user_service import UserService
from domain.errors import NotFoundError, ConflictError
from api.dependencies import get_user_service
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/{user_id}")
async def get_user(
user_id: str,
service: UserService = Depends(get_user_service),
):
try:
return await service.get_user(user_id)
except NotFoundError:
raise HTTPException(status_code=404, detail="User not found")Error Hierarchy
Define domain errors that are framework-agnostic. Map them to HTTP errors in the API layer.
# domain/errors.py
class DomainError(Exception):
"""Base class for all domain errors."""
def __init__(self, message: str):
self.message = message
super().__init__(message)
class NotFoundError(DomainError):
def __init__(self, resource: str, resource_id: str | int):
self.resource = resource
self.resource_id = resource_id
super().__init__(f"{resource} {resource_id} not found")
class ConflictError(DomainError):
pass
class ValidationError(DomainError):
def __init__(self, field: str, message: str):
self.field = field
super().__init__(f"Validation error on {field}: {message}")
class InactiveUserError(DomainError):
def __init__(self, user_id: str):
super().__init__(f"User {user_id} is inactive")
class OrderNotEditableError(DomainError):
def __init__(self, order_id: str, status: str):
super().__init__(f"Order {order_id} cannot be edited in status '{status}'")
class EmptyOrderError(DomainError):
def __init__(self, order_id: str):
super().__init__(f"Order {order_id} cannot be submitted with no items")Mapping Domain Errors to HTTP
# api/error_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from domain.errors import DomainError, NotFoundError, ConflictError, ValidationError
def register_error_handlers(app: FastAPI) -> None:
@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
return JSONResponse(status_code=404, content={"error": exc.message})
@app.exception_handler(ConflictError)
async def conflict_handler(request: Request, exc: ConflictError):
return JSONResponse(status_code=409, content={"error": exc.message})
@app.exception_handler(ValidationError)
async def validation_handler(request: Request, exc: ValidationError):
return JSONResponse(
status_code=422,
content={"error": exc.message, "field": exc.field},
)
@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
return JSONResponse(status_code=400, content={"error": exc.message})Testing Strategy
Unit Test Services with Mock Repos
# tests/unit/test_user_service.py
import pytest
from unittest.mock import AsyncMock
from domain.services.user_service import UserService
from domain.models.user import User, UserRole
from domain.errors import NotFoundError, ConflictError
@pytest.fixture
def mock_repo():
return AsyncMock()
@pytest.fixture
def service(mock_repo):
return UserService(user_repo=mock_repo)
@pytest.mark.asyncio
async def test_create_user_success(service, mock_repo):
mock_repo.get_by_email.return_value = None # No existing user
user = await service.create_user(email="new@test.com", name="New User")
assert user.email == "new@test.com"
assert user.role == UserRole.MEMBER
mock_repo.save.assert_called_once()
@pytest.mark.asyncio
async def test_create_user_duplicate_email(service, mock_repo):
mock_repo.get_by_email.return_value = User(
id="existing", email="dupe@test.com", name="Existing",
role=UserRole.MEMBER, created_at=utcnow(),
)
with pytest.raises(ConflictError, match="already exists"):
await service.create_user(email="dupe@test.com", name="New")
@pytest.mark.asyncio
async def test_promote_inactive_user_fails(service, mock_repo):
inactive_user = User(
id="u1", email="x@test.com", name="X",
role=UserRole.MEMBER, created_at=utcnow(), is_active=False,
)
mock_repo.get_by_id.return_value = inactive_user
with pytest.raises(InactiveUserError):
await service.promote_to_admin("u1")Integration Test Repos with Test DB
Use an in-memory SQLite database. Create tables in a fixture, inject the session into the repo, and test round-trip persistence.
# tests/integration/test_user_repo.py
@pytest.fixture
async def session():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with async_sessionmaker(engine)() as session:
yield session
@pytest.mark.asyncio
async def test_save_and_retrieve(session):
repo = SqlAlchemyUserRepository(session)
user = User(id="u1", email="test@test.com", name="Test",
role=UserRole.MEMBER, created_at=utcnow())
await repo.save(user)
await session.commit()
retrieved = await repo.get_by_id("u1")
assert retrieved is not None
assert retrieved.email == "test@test.com"Anti-patterns
Business Logic in Route Handlers
# BAD: Business logic mixed with HTTP handling
@router.post("/orders")
async def create_order(body: CreateOrderRequest, db = Depends(get_db)):
user = await db.get(User, body.user_id)
if not user.is_active:
raise HTTPException(400, "Inactive user") # Business rule in handler
if len(body.items) == 0:
raise HTTPException(400, "Empty order") # Business rule in handler
order = Order(...)
db.add(order)
# This is untestable without spinning up FastAPI
# GOOD: Handler delegates to service
@router.post("/orders")
async def create_order(body: CreateOrderRequest, service = Depends(get_order_service)):
return await service.create_order(body.user_id, body.items)ORM Models as Domain Models
Using SQLAlchemy models directly in business logic couples your domain to the database schema. Change a column name and your business logic breaks.
Importing Framework in Service Layer
# BAD: Service imports FastAPI
from fastapi import HTTPException
class UserService:
async def get_user(self, id):
user = await self.repo.get(id)
if not user:
raise HTTPException(404) # Framework leak!
# GOOD: Service raises domain error
from domain.errors import NotFoundError
class UserService:
async def get_user(self, id):
user = await self.repo.get(id)
if not user:
raise NotFoundError("User", id) # Framework-agnosticNo Error Hierarchy
Using bare Exception or ValueError everywhere makes it impossible to map domain errors to HTTP status codes consistently. Define a clear hierarchy rooted in DomainError.
Untestable Code
If you cannot test a service without starting a web server or connecting to a database, your architecture is wrong. Services should accept repository interfaces, and tests should inject mocks.