Clean Architecture
Clean Architecture is a software design philosophy that emphasizes separation of concerns, dependency inversion, and maintainability. It organizes code into concentric layers where inner layers contain business logic and outer layers handle implementation details like databases, web frameworks, and external services.
Core Principles
Dependency Rule: Dependencies point inward. Outer layers depend on inner layers, never the reverse. This ensures that business logic remains independent of implementation details.
Layer Independence: Each layer has a single responsibility and can be modified without affecting other layers, provided interfaces remain stable.
Framework Independence: The architecture doesn’t depend on specific frameworks or libraries. These are tools that serve the application, not the other way around.
Architecture Layers
┌─────────────────────────────────────┐
│ External Interfaces │
│ (Web, Database, File System) │
├─────────────────────────────────────┤
│ Interface Adapters │
│ (Controllers, Presenters, Gateways) │
├─────────────────────────────────────┤
│ Use Cases │
│ (Application Business Rules) │
├─────────────────────────────────────┤
│ Entities │
│ (Enterprise Business Rules) │
└─────────────────────────────────────┘
Entities: Core business objects containing enterprise-wide business rules. These change least frequently.
Use Cases: Application-specific business rules that orchestrate data flow between entities and external interfaces.
Interface Adapters: Convert data between use cases and external agencies. Include controllers, presenters, and gateways.
External Interfaces: Frameworks, databases, web servers, and other external tools.
Benefits
Testability: Business logic can be tested without databases, web servers, or external dependencies.
Flexibility: Easy to swap implementations (e.g., change from MySQL to PostgreSQL) without affecting business logic.
Maintainability: Clear separation makes code easier to understand and modify.
Technology Independence: Business rules don’t depend on specific technologies, making the system more resilient to technological changes.
When to Use
Clean Architecture works best for:
- Complex business applications with evolving requirements
- Systems requiring high testability and maintainability
- Applications that need to integrate with multiple external services
- Long-term projects where technology choices may change
It may be overkill for simple CRUD applications or small utilities where the overhead of multiple layers doesn’t provide sufficient value.
Example 1: User Management System
This example demonstrates a simple user management system following Clean Architecture principles.
from abc import ABC, abstractmethod
from typing import Optional, List
from dataclasses import dataclass
from datetime import datetime
# ================== ENTITIES LAYER ==================
@dataclass
class User:
"""Core business entity representing a user"""
user_id: str
email: str
name: str
created_at: datetime
is_active: bool = True
def deactivate(self) -> None:
"""Business rule: deactivate user"""
self.is_active = False
def is_valid_email(self) -> bool:
"""Business rule: validate email format"""
return "@" in self.email and "." in self.email
# ================== USE CASES LAYER ==================
class UserRepository(ABC):
"""Abstract interface for user data access"""
@abstractmethod
def save(self, user: User) -> None:
pass
@abstractmethod
def find_by_email(self, email: str) -> Optional[User]:
pass
class CreateUserUseCase:
"""Application business logic for creating users"""
def __init__(self, user_repository: UserRepository):
self._user_repository = user_repository
def execute(self, email: str, name: str) -> User:
"""Create a new user following business rules"""
# Check if user already exists
existing_user = self._user_repository.find_by_email(email)
if existing_user:
raise ValueError("User with this email already exists")
# Create new user
user = User(
user_id=f"user_{datetime.now().timestamp()}",
email=email,
name=name,
created_at=datetime.now()
)
# Validate business rules
if not user.is_valid_email():
raise ValueError("Invalid email format")
# Save user
self._user_repository.save(user)
return user
# ================== INTERFACE ADAPTERS LAYER ==================
class InMemoryUserRepository(UserRepository):
"""Concrete implementation of user repository using in-memory storage"""
def __init__(self):
self._users: List[User] = []
def save(self, user: User) -> None:
# Remove existing user with same email if any
self._users = [u for u in self._users if u.email != user.email]
self._users.append(user)
def find_by_email(self, email: str) -> Optional[User]:
for user in self._users:
if user.email == email:
return user
return None
class UserController:
"""HTTP controller handling user-related requests"""
def __init__(self, create_user_use_case: CreateUserUseCase):
self._create_user_use_case = create_user_use_case
def create_user(self, request_data: dict) -> dict:
"""Handle user creation request"""
try:
email = request_data.get("email")
name = request_data.get("name")
if not email or not name:
return {"error": "Email and name are required", "status": 400}
user = self._create_user_use_case.execute(email, name)
return {
"user_id": user.user_id,
"email": user.email,
"name": user.name,
"status": 201
}
except ValueError as e:
return {"error": str(e), "status": 400}
# ================== EXTERNAL INTERFACES LAYER ==================
# This would typically contain web framework code, database drivers, etc.
# For demonstration, we'll show how components are wired together
def main():
"""Dependency injection and application setup"""
# Create dependencies (outer layers)
user_repository = InMemoryUserRepository()
# Create use cases (inner layers)
create_user_use_case = CreateUserUseCase(user_repository)
# Create controllers (interface adapters)
user_controller = UserController(create_user_use_case)
# Simulate HTTP request
request = {"email": "john@example.com", "name": "John Doe"}
response = user_controller.create_user(request)
print(f"Response: {response}")
if __name__ == "__main__":
main()
This example shows how Clean Architecture separates concerns:
- Entities (
User) contain core business logic - Use Cases (
CreateUserUseCase) orchestrate business operations - Interface Adapters (
UserController,InMemoryUserRepository) handle external communication - Dependencies point inward - the use case depends on the abstract
UserRepository, not the concrete implementation
Example 2: Order Processing System
This example demonstrates Clean Architecture in an e-commerce order processing context.
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from decimal import Decimal
from enum import Enum
# ================== ENTITIES LAYER ==================
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
@dataclass
class OrderItem:
"""Value object representing an item in an order"""
product_id: str
quantity: int
unit_price: Decimal
def total_price(self) -> Decimal:
"""Calculate total price for this item"""
return self.unit_price * self.quantity
@dataclass
class Order:
"""Core business entity representing an order"""
order_id: str
customer_id: str
items: List[OrderItem]
status: OrderStatus = OrderStatus.PENDING
def total_amount(self) -> Decimal:
"""Business rule: calculate total order amount"""
return sum(item.total_price() for item in self.items)
def can_be_shipped(self) -> bool:
"""Business rule: check if order can be shipped"""
return self.status == OrderStatus.CONFIRMED
def ship(self) -> None:
"""Business rule: ship the order"""
if not self.can_be_shipped():
raise ValueError("Order must be confirmed before shipping")
self.status = OrderStatus.SHIPPED
# ================== USE CASES LAYER ==================
class OrderRepository(ABC):
"""Abstract interface for order data access"""
@abstractmethod
def save(self, order: Order) -> None:
pass
@abstractmethod
def find_by_id(self, order_id: str) -> Optional[Order]:
pass
class PaymentGateway(ABC):
"""Abstract interface for payment processing"""
@abstractmethod
def process_payment(self, amount: Decimal, customer_id: str) -> bool:
pass
class ProcessOrderUseCase:
"""Application business logic for processing orders"""
def __init__(self, order_repository: OrderRepository,
payment_gateway: PaymentGateway):
self._order_repository = order_repository
self._payment_gateway = payment_gateway
def execute(self, order_id: str) -> Order:
"""Process an order by handling payment and updating status"""
# Retrieve order
order = self._order_repository.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
if order.status != OrderStatus.PENDING:
raise ValueError("Order is not in pending status")
# Process payment
payment_successful = self._payment_gateway.process_payment(
order.total_amount(),
order.customer_id
)
if payment_successful:
order.status = OrderStatus.CONFIRMED
self._order_repository.save(order)
else:
raise ValueError("Payment processing failed")
return order
# ================== INTERFACE ADAPTERS LAYER ==================
class InMemoryOrderRepository(OrderRepository):
"""Concrete implementation using in-memory storage"""
def __init__(self):
self._orders: dict[str, Order] = {}
def save(self, order: Order) -> None:
self._orders[order.order_id] = order
def find_by_id(self, order_id: str) -> Optional[Order]:
return self._orders.get(order_id)
class MockPaymentGateway(PaymentGateway):
"""Mock payment gateway for testing"""
def process_payment(self, amount: Decimal, customer_id: str) -> bool:
# Simulate payment processing logic
# In real implementation, this would call external payment service
return amount > Decimal('0') and customer_id is not None
class OrderController:
"""Controller handling order-related operations"""
def __init__(self, process_order_use_case: ProcessOrderUseCase):
self._process_order_use_case = process_order_use_case
def process_order(self, order_id: str) -> dict:
"""Handle order processing request"""
try:
order = self._process_order_use_case.execute(order_id)
return {
"order_id": order.order_id,
"status": order.status.value,
"total_amount": str(order.total_amount()),
"success": True
}
except ValueError as e:
return {
"error": str(e),
"success": False
}
# ================== EXTERNAL INTERFACES LAYER ==================
def setup_application():
"""Dependency injection setup"""
# Create infrastructure components
order_repository = InMemoryOrderRepository()
payment_gateway = MockPaymentGateway()
# Create sample order
sample_order = Order(
order_id="ORD-001",
customer_id="CUST-123",
items=[
OrderItem("PROD-1", 2, Decimal('29.99')),
OrderItem("PROD-2", 1, Decimal('15.50'))
]
)
order_repository.save(sample_order)
# Wire up use cases
process_order_use_case = ProcessOrderUseCase(order_repository, payment_gateway)
# Create controller
order_controller = OrderController(process_order_use_case)
return order_controller, sample_order.order_id
def main():
"""Demonstrate the order processing system"""
controller, order_id = setup_application()
print(f"Processing order: {order_id}")
result = controller.process_order(order_id)
print(f"Result: {result}")
if __name__ == "__main__":
main()
This example illustrates key Clean Architecture concepts:
- Business Logic Isolation: Order rules (total calculation, shipping validation) are in the Entity layer
- Dependency Inversion: Use cases depend on abstract interfaces (
OrderRepository,PaymentGateway), not concrete implementations - Testability: Business logic can be tested independently of external systems
- Flexibility: Payment gateway or repository implementations can be swapped without changing business logic
The architecture ensures that core business rules remain stable while allowing external concerns to evolve independently.