Clean Architecture in Python: A Practical Guide
If you've ever inherited a codebase where business logic is tangled with database queries and HTTP handlers, you know the pain. Clean Architecture offers a way out—a set of principles that keep your code organized as it grows.
This guide walks through implementing Clean Architecture in Python with practical examples. We'll build a real structure you can adapt for your own projects, covering the layers, dependency rules, and common pitfalls I've encountered in production systems.
Problem
Most Python projects start simple. A Flask or FastAPI app with routes that call the database directly. It works fine for small apps. But as the codebase grows, problems emerge:
- Testing becomes difficult — Your business logic is mixed with framework code, so unit tests require mocking HTTP contexts and database connections
- Changing frameworks is painful — Switching from Flask to FastAPI means rewriting most of your code
- Database migrations break everything — A schema change forces updates across multiple layers
- New developers struggle — There's no clear place for different types of code
These issues compound over time. What started as a clean project becomes a tangled mess where a simple feature takes days to implement because you're afraid of breaking something else.
Why This Matters
Clean Architecture solves these problems by establishing clear boundaries. The core insight is that business rules shouldn't depend on frameworks, databases, or external services. Instead, dependencies should point inward—outer layers depend on inner layers, never the reverse.
This matters because:
- Your business logic becomes testable in isolation — No database mocks, no HTTP context
- Frameworks become replaceable — FastAPI today, something else tomorrow
- The codebase scales — New features go in predictable places
- Onboarding improves — Developers understand where to look
NOTE: Clean Architecture adds initial complexity. For small scripts or throwaway prototypes, keep it simple. But for anything that needs to be maintained, the investment pays off.
Solution
Clean Architecture organizes code into concentric layers. Each layer has specific responsibilities and follows the dependency rule: code can only depend on layers closer to the center.
The Layers
Entities (Domain Layer) — Core business objects and rules. No dependencies on anything external.
Use Cases (Application Layer) — Application-specific business rules. Orchestrates entities and defines interfaces for external dependencies.
Interface Adapters — Converts data between the format used by use cases and the format used by external agencies like databases or web frameworks.
Frameworks & Drivers — The outermost layer. Frameworks, database drivers, web servers, etc.
Implementation
Let's build a user management system to demonstrate these concepts.
1. Entities (Domain Layer)
Entities are pure Python objects representing core business concepts. They contain business logic but no framework dependencies.
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import re
@dataclass
class Email:
"""Value object for email addresses with validation."""
value: str
def __post_init__(self):
if not self._is_valid(self.value):
raise ValueError(f"Invalid email format: {self.value}")
@staticmethod
def _is_valid(email: str) -> bool:
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}#39;
return bool(re.match(pattern, email))
def __str__(self) -> str:
return self.value
@dataclass
class User:
"""Core user entity with business rules."""
id: str
email: Email
name: str
is_active: bool = True
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: Optional[datetime] = None
def deactivate(self) -> None:
"""Deactivates the user account."""
self.is_active = False
self.updated_at = datetime.utcnow()
def update_email(self, new_email: str) -> None:
"""Updates user email with validation."""
self.email = Email(new_email)
self.updated_at = datetime.utcnow()
def can_perform_action(self) -> bool:
"""Check if user can perform protected actions."""
return self.is_active
TIP: Use value objects like
2. Use Cases (Application Layer)
Use cases contain application-specific business rules. They define what the system does, not how it does it.
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
# Repository interface - defined in application layer, implemented in infrastructure
class UserRepository(ABC):
@abstractmethod
async def get_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
async def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def delete(self, user_id: str) -> bool:
pass
# Use case input/output DTOs
@dataclass
class CreateUserRequest:
email: str
name: str
@dataclass
class CreateUserResponse:
id: str
email: str
name: str
created_at: datetime
# Use case implementation
class CreateUserUseCase:
def __init__(self, user_repository: UserRepository):
self._repo = user_repository
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
# Check if email already exists
existing = await self._repo.get_by_email(request.email)
if existing:
raise UserAlreadyExistsError(f"User with email {request.email} already exists")
# Create domain entity
user = User(
id=generate_uuid(),
email=Email(request.email),
name=request.name
)
# Persist and return
saved_user = await self._repo.save(user)
return CreateUserResponse(
id=saved_user.id,
email=str(saved_user.email),
name=saved_user.name,
created_at=saved_user.created_at
)
WARNING: Don't let framework code leak into use cases. If you see
Request,Response, or ORM models here, you're breaking the architecture.
3. Interface Adapters
This layer converts data between the use case format and external formats.
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, EmailStr
router = APIRouter(prefix="/users", tags=["users"])
# Pydantic models for API
class CreateUserAPIRequest(BaseModel):
email: EmailStr
name: str
class Config:
json_schema_extra = {
"example": {
"email": "user@example.com",
"name": "John Doe"
}
}
class UserAPIResponse(BaseModel):
id: str
email: str
name: str
created_at: datetime
class Config:
from_attributes = True
# Route handler - converts between API format and use case format
@router.post("/", response_model=UserAPIResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
request: CreateUserAPIRequest,
use_case: CreateUserUseCase = Depends(get_create_user_use_case)
):
try:
# Convert API request to use case request
use_case_request = CreateUserRequest(
email=request.email,
name=request.name
)
# Execute use case
result = await use_case.execute(use_case_request)
# Convert use case response to API response
return UserAPIResponse(
id=result.id,
email=result.email,
name=result.name,
created_at=result.created_at
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
4. Frameworks & Drivers
The outermost layer contains framework-specific implementations.
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
# SQLAlchemy ORM model
class UserModel(Base):
__tablename__ = "users"
id = Column(String, primary_key=True)
email = Column(String, unique=True, nullable=False, index=True)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, nullable=True)
# Repository implementation
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: AsyncSession):
self._session = session
async def get_by_id(self, user_id: str) -> Optional[User]:
result = await self._session.execute(
select(UserModel).where(UserModel.id == user_id)
)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def get_by_email(self, email: str) -> Optional[User]:
result = await self._session.execute(
select(UserModel).where(UserModel.email == email)
)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def save(self, user: User) -> User:
model = UserModel(
id=user.id,
email=str(user.email),
name=user.name,
is_active=user.is_active,
created_at=user.created_at,
updated_at=user.updated_at
)
self._session.add(model)
await self._session.commit()
return user
def _to_entity(self, model: UserModel) -> User:
return User(
id=model.id,
email=Email(model.email),
name=model.name,
is_active=model.is_active,
created_at=model.created_at,
updated_at=model.updated_at
)
Example: Directory Structure
Here's how the full project structure looks:
src/
├── domain/ # Entities layer
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ └── email.py
│ └── exceptions.py
│
├── application/ # Use cases layer
│ ├── __init__.py
│ ├── interfaces/
│ │ ├── __init__.py
│ │ └── repositories.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── get_user.py
│ └── dtos.py
│
├── infrastructure/ # Frameworks & drivers
│ ├── __init__.py
│ ├── database/
│ │ ├── __init__.py
│ │ ├── models.py
│ │ └── repositories.py
│ └── external/
│ └── email_service.py
│
├── presentation/ # Interface adapters
│ ├── __init__.py
│ └── api/
│ ├── __init__.py
│ ├── routes/
│ │ └── users.py
│ └── dependencies.py
│
└── main.py # Entry point and DI wiring
Common Mistakes
1. Leaking ORM Models into Use Cases
# Wrong - ORM model in use case
class CreateUserUseCase:
async def execute(self, user_model: UserModel): # ORM dependency!
pass
# Correct - Domain entity in use case
class CreateUserUseCase:
async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
pass
2. Implementing Interfaces in the Wrong Layer
Repository interfaces belong in the Application layer, not Infrastructure. The Application layer defines what it needs, Infrastructure provides it.
3. Skipping the Interface Adapter Layer
It's tempting to call use cases directly from route handlers with framework types. This couples your business logic to the framework and makes testing harder.
4. Over-Engineering for Simple Projects
Clean Architecture shines in complex domains. For a simple CRUD API with minimal business logic, it adds unnecessary complexity. Start simple and refactor when needed.
Conclusion
Clean Architecture isn't about following rules for their own sake—it's about creating systems that are testable, maintainable, and adaptable to change. The dependency rule (inward dependencies only) is the key insight.
Start with clear layer boundaries between domain logic and infrastructure. Use interfaces to define contracts between layers. Test your business logic in isolation. Your future self (and your team) will thank you.
The upfront investment is real, but so is the payoff: a codebase that's actually pleasant to work with as it grows.