Clean Architecture is a software design philosophy that emphasizes:
- Independence of Frameworks: Business logic is not coupled to frameworks or libraries
- Testability: Core business logic can be tested without UI, database, or external services
- Independence of UI: The UI can change without affecting business logic
- Independence of Database: Database choice is an implementation detail
- Independence of External Agencies: Business rules don't depend on external APIs or services
Source code dependencies must point inward. Nothing in an inner circle can know anything about outer circles.
This means:
- ✅ Outer layers can depend on inner layers
- ❌ Inner layers MUST NOT depend on outer layers
- ❌ No circular dependencies
Location: src/domain/
The core of the application, containing pure business logic with no external dependencies.
Components:
-
Entities: Core business objects representing concepts in your domain
- Example:
User,Product,Order - Contains business rules and invariants
- Should be framework-agnostic
- Example:
-
Value Objects: Immutable objects defined by their values, not identity
- Example:
Email,Money,Address - Encapsulate validation logic
- Example:
-
Repository Interfaces: Abstract contracts for data access
- Define WHAT data operations are needed (not HOW)
- Enable testing with mock repositories
- Allow switching database implementations
-
Domain Exceptions: Custom exceptions for business rule violations
- Example:
UserAlreadyExistsError,InsufficientFundsError
- Example:
Characteristics:
- ✅ Zero external dependencies
- ✅ Pure Python code
- ✅ Highly testable
- ✅ Reusable in different contexts (CLI, web, batch jobs)
Example:
# Domain layer - pure business logic
class User:
def __init__(self, id: int, email: str, name: str):
if not self._is_valid_email(email):
raise InvalidEmailError()
self.id = id
self.email = email
self.name = name
@staticmethod
def _is_valid_email(email: str) -> bool:
return "@" in emailLocation: src/application/
Orchestrates domain objects and implements use cases. It's the "glue" between domain logic and external concerns.
Components:
-
Use Cases / Application Services: Define application-specific business rules
- Example:
CreateUserUseCase,UpdateProductPriceUseCase - Orchestrate domain objects
- Coordinate with repositories
- Handle transactions and domain events
- Example:
-
DTOs (Data Transfer Objects): Simplified data structures for transferring between layers
- Input DTOs: Carry data into use cases
- Output DTOs: Carry results out of use cases
-
Mappers: Convert between domain entities and DTOs
- Decouple internal representation from external communication
Characteristics:
- ✅ Depends only on domain layer
- ✅ No knowledge of frameworks
- ✅ No direct database access (uses repository interfaces)
- ✅ Handles use case orchestration
- ✅ Highly testable with mock repositories
Example:
# Application layer - use case orchestration
class CreateUserUseCase:
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository
async def execute(self, email: str, name: str) -> UserOutputDTO:
# Check business rule: email uniqueness
existing_user = await self.user_repository.find_by_email(email)
if existing_user:
raise UserAlreadyExistsError(email)
# Create domain entity
user = User(id=None, email=email, name=name)
# Persist through repository interface
created_user = await self.user_repository.save(user)
# Return DTO
return UserOutputDTO.from_entity(created_user)Location: src/infrastructure/
Implements the concrete details of data access, external services, and other technical concerns.
Components:
-
Repository Implementations: Concrete implementations of repository interfaces
- Example:
SQLAlchemyUserRepository,MongoDBUserRepository - Handle database queries and persistence
- Manage ORM/ODM models
- Example:
-
Database Models: ORM/ODM representations of domain entities
- Example: SQLAlchemy models
- Separate from domain entities
- Framework-specific concerns
-
External Service Adapters: Integrate with external APIs
- Example:
EmailServiceAdapter,PaymentGatewayAdapter - Translate external APIs to domain concepts
- Example:
-
Configuration: Database connections, API clients
- Initialization of dependencies
Characteristics:
- ✅ Implements repository interfaces from domain
- ✅ Handles database-specific logic
- ✅ Contains framework-specific code (SQLAlchemy, ORMs)
- ✅ Can have external dependencies
- ✅ Easily replaceable implementations
Example:
# Infrastructure layer - concrete database implementation
class SQLAlchemyUserRepository(IUserRepository):
def __init__(self, session: AsyncSession):
self.session = session
async def find_by_email(self, email: str) -> User | None:
# Database-specific query
result = await self.session.execute(
select(UserModel).where(UserModel.email == email)
)
user_model = result.scalar_one_or_none()
if not user_model:
return None
# Map ORM model back to domain entity
return self._to_domain(user_model)
async def save(self, user: User) -> User:
# Convert domain entity to ORM model
user_model = self._to_model(user)
self.session.add(user_model)
await self.session.commit()
return self._to_domain(user_model)
def _to_domain(self, model: UserModel) -> User:
return User(id=model.id, email=model.email, name=model.name)
def _to_model(self, user: User) -> UserModel:
return UserModel(id=user.id, email=user.email, name=user.name)Location: src/presentation/
Handles HTTP communication, request validation, and API formatting. FastAPI specific code.
Components:
-
FastAPI Routers: Define HTTP endpoints
- Handle HTTP methods (GET, POST, PUT, DELETE)
- Define URL paths
- Specify response codes and descriptions
-
Request/Response Schemas: Pydantic models for validation
- Validate incoming request data
- Serialize outgoing response data
- Document API in OpenAPI
-
Dependencies: FastAPI dependency injection
- Provide repositories and use cases to routers
- Handle database sessions
- Manage transient vs singleton lifecycles
-
Middleware: Cross-cutting concerns
- Error handling
- Logging
- CORS, authentication, authorization
Characteristics:
- ✅ FastAPI-specific code
- ✅ Depends on application and domain layers
- ✅ Handles HTTP protocol details
- ✅ Validates and formats data
Example:
# Presentation layer - FastAPI routers
from fastapi import APIRouter, Depends, HTTPException
router = APIRouter(prefix="/api/users", tags=["users"])
async def get_create_user_use_case(
db_session: AsyncSession = Depends(get_db_session)
) -> CreateUserUseCase:
repository = SQLAlchemyUserRepository(db_session)
return CreateUserUseCase(repository)
@router.post("/", status_code=201)
async def create_user(
request: CreateUserRequest,
use_case: CreateUserUseCase = Depends(get_create_user_use_case)
) -> UserResponse:
try:
result = await use_case.execute(
email=request.email,
name=request.name
)
return UserResponse(**result.dict())
except UserAlreadyExistsError as e:
raise HTTPException(status_code=409, detail=str(e))1. HTTP Request arrives at FastAPI router
POST /api/users { "email": "user@example.com", "name": "John" }
↓
2. Request Schema validates the input (Pydantic)
↓
3. Router receives dependencies (repository, use case)
↓
4. Use Case is executed (application layer)
- Checks business rule: unique email
- Creates User entity (domain layer)
- Calls repository to persist
↓
5. Repository saves to database (infrastructure layer)
- Converts User entity to ORM model
- Executes SQL query
- Returns persisted entity
↓
6. Use Case maps result to DTO
↓
7. Router converts DTO to Response Schema
↓
8. FastAPI serializes to JSON and returns HTTP 201
{ "id": 1, "email": "user@example.com", "name": "John" }
Abstracts data access, allowing domain logic to remain agnostic about persistence implementation.
# Domain defines interface
class IUserRepository(ABC):
async def find_by_email(self, email: str) -> User | None: ...
async def save(self, user: User) -> User: ...
# Infrastructure provides implementation
class SQLAlchemyUserRepository(IUserRepository):
async def find_by_email(self, email: str) -> User | None: ...Loose coupling between components. Dependencies are injected rather than created internally.
# Use case doesn't create repository; it's injected
class CreateUserUseCase:
def __init__(self, user_repository: IUserRepository):
self.user_repository = user_repository # Injected, not createdSimplifies data transfer between layers, preventing domain model leakage.
# Application layer returns DTO, not domain entity
class UserOutputDTO:
id: int
email: str
name: strConverts between domain entities and other representations.
class UserMapper:
@staticmethod
def to_domain(user_model: UserModel) -> User: ...
@staticmethod
def to_response(user: User) -> UserResponse: ...Each class has one reason to change. Use cases orchestrate, repositories persist, entities encapsulate rules.
Open for extension (add new repositories), closed for modification (don't change existing domain rules).
Repository implementations are interchangeable. You can swap SQLAlchemy for MongoDB without affecting use cases.
Narrow, focused interfaces. IUserRepository doesn't include product-related methods.
Depend on abstractions, not concretions. Use cases depend on IUserRepository, not SQLAlchemyUserRepository.
✅ Testability: Domain logic and use cases tested without database ✅ Maintainability: Clear structure, easy to find where to make changes ✅ Flexibility: Swap databases, frameworks without touching business logic ✅ Scalability: Easy to add new features without affecting existing code ✅ Reusability: Domain logic can be used in CLI, scheduled jobs, batch processing ✅ Team Collaboration: Clear boundaries allow developers to work independently ✅ Onboarding: New developers understand the structure quickly
For very small prototypes or scripts, this architecture may be overkill. Use judgment based on project scope.
Further Reading:
- "Clean Architecture" by Robert C. Martin
- Domain-Driven Design patterns
- SOLID principles documentation