Clean Architecture Guide - Noloquideus/fastapi-template GitHub Wiki
Clean Architecture in FastAPI: A Comprehensive Guide
Table of Contents
- Introduction
- Core Principles
- Architecture Layers
- Project Structure
- Layer Dependencies
- Implementation Examples
- Benefits
- Best Practices
- Common Pitfalls
Introduction
Clean Architecture, introduced by Robert C. Martin (Uncle Bob), is a software design philosophy that emphasizes the separation of concerns and independence of frameworks, databases, and external agencies. This FastAPI template demonstrates a practical implementation of Clean Architecture principles in a modern Python web application.
The main goal is to create a system that is:
- Independent of frameworks: The architecture doesn't depend on specific frameworks
- Testable: Business rules can be tested without UI, databases, or external elements
- Independent of UI: The UI can change without changing the business rules
- Independent of database: Business rules are not bound to the database
- Independent of external agencies: Business rules don't know anything about interfaces to external systems
Core Principles
1. Dependency Inversion
Dependencies point inward toward higher-level policies. Outer layers depend on inner layers, never the reverse.
2. Single Responsibility
Each layer has a single, well-defined responsibility and reason to change.
3. Interface Segregation
No layer should be forced to depend on interfaces it doesn't use.
4. Open/Closed Principle
Layers should be open for extension but closed for modification.
Architecture Layers
Our FastAPI template implements three main layers:
┌─────────────────────────────────────────┐
│ Presentation Layer │ ← External interfaces
│ (Controllers, Middleware) │
├─────────────────────────────────────────┤
│ Application Layer │ ← Business logic & use cases
│ (Services, Abstractions) │
├─────────────────────────────────────────┤
│ Domain Layer │ ← Core business rules
│ (Entities, Value Objects, Rules) │
└─────────────────────────────────────────┘
↑
Infrastructure Layer (Database, External APIs)
Domain Layer (Core)
The innermost layer containing:
- Entities: Core business objects
- Value Objects: Immutable objects with no identity
- Domain Services: Business logic that doesn't belong to entities
- Domain Events: Important business events
Application Layer
Contains:
- Use Cases: Application-specific business rules
- Interfaces: Abstractions for external dependencies
- DTOs: Data transfer objects
- Application Services: Orchestration of domain objects
Presentation Layer
Handles:
- Controllers: HTTP request/response handling
- Middleware: Cross-cutting concerns
- Serialization: Data format conversion
- Authentication: Security concerns
Infrastructure Layer
Implements:
- Repositories: Data access implementations
- External Services: Third-party integrations
- Frameworks: Database, messaging, etc.
Project Structure
src/
├── application/ # Application Layer
│ ├── abstractions/ # Interfaces and abstract classes
│ ├── contracts/ # Data contracts and DTOs
│ ├── domain/ # Domain Layer
│ │ ├── entities/ # Business entities
│ │ ├── exceptions/ # Domain exceptions
│ │ ├── objects/ # Value objects and domain objects
│ │ └── enums/ # Domain enumerations
│ └── services/ # Application services
├── infrastructure/ # Infrastructure Layer
│ ├── database/ # Database implementations
│ ├── logger/ # Logging infrastructure
│ └── utils/ # Utility functions
└── presentation/ # Presentation Layer
├── handlers/ # Exception handlers
├── middleware/ # HTTP middleware
├── routing/ # API routes
└── schemas/ # Request/Response schemas
Layer Dependencies
The dependency flow follows the Clean Architecture rules:
Presentation → Application → Domain
↑ ↑
Infrastructure ─────┘
- Domain: No dependencies on other layers
- Application: Depends only on Domain
- Presentation: Depends on Application and Domain
- Infrastructure: Depends on Application and Domain
Implementation Examples
Domain Layer Example
# src/application/domain/objects/immutable.py
from typing import Any
from src.application.domain.exceptions import ImmutableAttributeError
class Immutable:
"""
Base class for immutable objects.
Ensures object consistency after creation.
"""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self._frozen = False
super().__init__(*args, **kwargs)
self._frozen = True
def __setattr__(self, key: str, value: Any) -> None:
if hasattr(self, '_frozen') and self._frozen:
raise ImmutableAttributeError(
f"Cannot modify attribute '{key}' of immutable object"
)
super().__setattr__(key, value)
Application Layer Example
# src/application/services/user_service.py
from abc import ABC, abstractmethod
from typing import List
from src.application.domain.entities.user import User
class UserRepository(ABC):
@abstractmethod
async def create(self, user: User) -> User:
pass
@abstractmethod
async def find_by_email(self, email: str) -> User | None:
pass
class UserService:
def __init__(self, user_repository: UserRepository):
self._user_repository = user_repository
async def register_user(self, email: str, password: str) -> User:
# Business logic here
existing_user = await self._user_repository.find_by_email(email)
if existing_user:
raise ValueError("User already exists")
user = User(email=email, password=password)
return await self._user_repository.create(user)
Presentation Layer Example
# src/presentation/middleware/trace_id.py
import uuid
from typing import Callable
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
class TraceIDMiddleware(BaseHTTPMiddleware):
async def dispatch(
self,
request: Request,
call_next: Callable
) -> Response:
trace_id = str(uuid.uuid4())
request.state.trace_id = trace_id
response = await call_next(request)
response.headers["X-Trace-ID"] = trace_id
return response
Infrastructure Layer Example
# src/infrastructure/database/repositories/user_repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.application.services.user_service import UserRepository
from src.application.domain.entities.user import User
from src.infrastructure.database.models.user import UserModel
class SQLUserRepository(UserRepository):
def __init__(self, session: AsyncSession):
self._session = session
async def create(self, user: User) -> User:
user_model = UserModel(
email=user.email,
password=user.password
)
self._session.add(user_model)
await self._session.commit()
return user
async def find_by_email(self, email: str) -> User | None:
result = await self._session.execute(
select(UserModel).where(UserModel.email == email)
)
user_model = result.scalar_one_or_none()
if user_model:
return User(email=user_model.email, password=user_model.password)
return None
Benefits
1. Testability
Each layer can be tested in isolation:
# Testing application service with mocked repository
async def test_register_user():
mock_repository = Mock(spec=UserRepository)
mock_repository.find_by_email.return_value = None
user_service = UserService(mock_repository)
user = await user_service.register_user("[email protected]", "password")
assert user.email == "[email protected]"
mock_repository.create.assert_called_once()
2. Flexibility
Easy to swap implementations:
# Development: In-memory repository
user_service = UserService(InMemoryUserRepository())
# Production: Database repository
user_service = UserService(SQLUserRepository(session))
# Testing: Mock repository
user_service = UserService(MockUserRepository())
3. Maintainability
Clear separation of concerns makes the code easier to understand and modify.
4. Framework Independence
Business logic is not tied to FastAPI, making it easy to switch frameworks if needed.
Best Practices
1. Keep Domain Pure
# ✅ Good - Domain object with no external dependencies
class User:
def __init__(self, email: str, password: str):
self.email = email
self.password = self._hash_password(password)
def _hash_password(self, password: str) -> str:
# Pure domain logic
return password # Simplified
# ❌ Bad - Domain object depending on external library
class User:
def __init__(self, email: str, password: str):
import bcrypt # External dependency in domain
self.email = email
self.password = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
2. Use Dependency Injection
# ✅ Good - Dependencies injected
class UserService:
def __init__(self, user_repository: UserRepository, logger: Logger):
self._user_repository = user_repository
self._logger = logger
# ❌ Bad - Direct instantiation
class UserService:
def __init__(self):
self._user_repository = SQLUserRepository() # Tight coupling
self._logger = Logger() # Hard to test
3. Define Clear Interfaces
# ✅ Good - Abstract interface
from abc import ABC, abstractmethod
class EmailService(ABC):
@abstractmethod
async def send_email(self, to: str, subject: str, body: str) -> None:
pass
# ❌ Bad - Concrete implementation as interface
class SMTPEmailService:
async def send_email(self, to: str, subject: str, body: str) -> None:
# SMTP specific implementation
pass
4. Keep Controllers Thin
# ✅ Good - Controller delegates to service
@app.post("/users")
async def create_user(
request: CreateUserRequest,
user_service: UserService = Depends()
):
user = await user_service.register_user(request.email, request.password)
return UserResponse.from_entity(user)
# ❌ Bad - Business logic in controller
@app.post("/users")
async def create_user(request: CreateUserRequest):
# Validation logic
if not request.email or "@" not in request.email:
raise HTTPException(400, "Invalid email")
# Database logic
session = get_db_session()
existing = session.query(UserModel).filter_by(email=request.email).first()
if existing:
raise HTTPException(409, "User exists")
# More business logic...
Common Pitfalls
1. Circular Dependencies
Avoid importing from outer layers in inner layers:
# ❌ Bad - Domain importing from infrastructure
from src.infrastructure.database.models import UserModel # In domain layer
# ✅ Good - Use abstractions
from abc import ABC, abstractmethod
2. Anemic Domain Model
Ensure domain objects contain behavior, not just data:
# ❌ Bad - Anemic model
class User:
def __init__(self, email: str, password: str):
self.email = email
self.password = password
# ✅ Good - Rich domain model
class User:
def __init__(self, email: str, password: str):
self.email = self._validate_email(email)
self.password = password
self.is_active = True
def deactivate(self) -> None:
self.is_active = False
def change_password(self, new_password: str) -> None:
self.password = self._hash_password(new_password)
3. Over-Engineering
Don't create unnecessary abstractions for simple operations:
# ❌ Bad - Over-engineered for simple operation
class StringValidatorInterface(ABC):
@abstractmethod
def validate(self, value: str) -> bool:
pass
class EmailValidator(StringValidatorInterface):
def validate(self, value: str) -> bool:
return "@" in value
# ✅ Good - Simple validation
def is_valid_email(email: str) -> bool:
return "@" in email and "." in email
Conclusion
Clean Architecture provides a robust foundation for building maintainable, testable, and scalable applications. This FastAPI template demonstrates how to implement these principles in a practical way while avoiding common pitfalls.
The key is to:
- Keep dependencies pointing inward
- Separate concerns clearly
- Use abstractions for external dependencies
- Test each layer in isolation
- Keep the domain pure and framework-independent
By following these guidelines, you'll create applications that are easier to understand, test, and maintain over time.
Further Reading: