ποΈ Clean Architecture PatternΒΆ
Estimated reading time: 20 minutes
The Clean Architecture pattern enforces a layered approach where dependencies only flow inward, ensuring testability, maintainability, and independence from external concerns.
π― What & WhyΒΆ
The Problem: Tightly Coupled LayersΒΆ
Without clean architecture, code becomes tangled with business logic mixed with infrastructure concerns:
# β Problem: Business logic tightly coupled to framework and database
from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
import stripe
app = FastAPI()
mongo_client = MongoClient("mongodb://localhost:27017")
db = mongo_client.pizzeria
@app.post("/orders")
async def place_order(order_data: dict):
# β HTTP framework logic mixed with business logic
try:
# β Database details in endpoint handler
customer = db.customers.find_one({"_id": order_data["customer_id"]})
if not customer:
raise HTTPException(status_code=404, detail="Customer not found")
# β Business rules scattered in controller
subtotal = sum(item["price"] for item in order_data["items"])
tax = subtotal * 0.08
total = subtotal + tax
# β Direct payment API call in controller
stripe.api_key = "sk_test_..."
charge = stripe.Charge.create(
amount=int(total * 100),
currency="usd",
source=order_data["payment_token"]
)
# β Direct MongoDB operations
order_doc = {
"customer_id": order_data["customer_id"],
"items": order_data["items"],
"total": total,
"status": "pending",
"stripe_charge_id": charge.id
}
result = db.orders.insert_one(order_doc)
# β HTTP response mixed with business logic
return {
"order_id": str(result.inserted_id),
"total": total,
"status": "pending"
}
except stripe.error.CardError as e:
# β Infrastructure exceptions in business layer
raise HTTPException(status_code=402, detail=str(e))
except Exception as e:
# β Generic error handling
raise HTTPException(status_code=500, detail=str(e))
# β Testing requires real MongoDB and Stripe
# β Can't swap database without rewriting entire endpoint
# β Business logic can't be reused for CLI or mobile app
# β Framework upgrade requires changing business logic
Problems with this approach:
- No Testability: Can't test without real database and payment service
- Tight Coupling: Business logic depends on FastAPI, MongoDB, Stripe
- No Reusability: Can't use order placement logic in CLI or batch jobs
- Hard to Maintain: Changes to infrastructure affect business logic
- Framework Lock-in: Stuck with FastAPI, can't migrate to another framework
- No Business Focus: Domain rules lost in infrastructure code
The Solution: Clean Architecture with Layer SeparationΒΆ
Separate concerns into layers with clear dependency direction:
# β
Solution: Layer 1 - Domain (Core Business Logic)
# domain/entities/order.py
from neuroglia.data.abstractions import Entity
class Order(Entity):
"""Pure business logic - no framework dependencies"""
def __init__(self, customer_id: str, items: List[OrderItem]):
super().__init__()
self.customer_id = customer_id
self.items = items
self.status = OrderStatus.PENDING
self.total = self._calculate_total()
# β
Domain events for business occurrences
self.raise_event(OrderPlacedEvent(
order_id=self.id,
customer_id=customer_id,
total=self.total
))
def _calculate_total(self) -> Decimal:
"""β
Business rule encapsulated in entity"""
subtotal = sum(item.price * item.quantity for item in self.items)
tax = subtotal * Decimal('0.08') # 8% tax
return subtotal + tax
# β
Layer 2 - Application (Use Cases)
# application/handlers/place_order_handler.py
from neuroglia.mediation import CommandHandler
class PlaceOrderHandler(CommandHandler[PlaceOrderCommand, OperationResult[OrderDto]]):
"""Orchestrates use case - depends only on abstractions"""
def __init__(
self,
order_repository: IOrderRepository, # β
Interface, not implementation
payment_service: IPaymentService, # β
Interface, not Stripe directly
mapper: Mapper
):
self._repository = order_repository
self._payment = payment_service
self._mapper = mapper
async def handle_async(self, command: PlaceOrderCommand):
# β
Use domain entity (business logic)
order = Order(command.customer_id, command.items)
# β
Use abstraction (can swap implementations)
payment_result = await self._payment.process_async(
amount=order.total,
payment_method=command.payment_method
)
if not payment_result.success:
return self.bad_request("Payment failed")
# β
Use repository abstraction
await self._repository.save_async(order)
# β
Return DTO, not entity
return self.created(self._mapper.map(order, OrderDto))
# β
Layer 3 - API (External Interface)
# api/controllers/orders_controller.py
from neuroglia.mvc import ControllerBase
class OrdersController(ControllerBase):
"""Thin controller - no business logic"""
@post("/", response_model=OrderDto, status_code=201)
async def place_order(self, request: PlaceOrderRequest):
# β
Map HTTP request to command
command = self.mapper.map(request, PlaceOrderCommand)
# β
Delegate to mediator (no direct handler dependency)
result = await self.mediator.execute_async(command)
# β
Process result (handles error codes)
return self.process(result)
# β
Layer 4 - Integration (Infrastructure)
# integration/services/stripe_payment_service.py
class StripePaymentService(IPaymentService):
"""Implementation detail - can be swapped"""
async def process_async(
self,
amount: Decimal,
payment_method: str
) -> PaymentResult:
try:
charge = stripe.Charge.create(
amount=int(amount * 100),
currency="usd",
source=payment_method
)
return PaymentResult(success=True, transaction_id=charge.id)
except stripe.error.CardError as e:
return PaymentResult(success=False, error=str(e))
# integration/repositories/mongo_order_repository.py
class MongoOrderRepository(IOrderRepository):
"""Implementation detail - can be swapped"""
async def save_async(self, order: Order) -> None:
doc = self._entity_to_document(order)
await self._collection.insert_one(doc)
Benefits of clean architecture:
- Testability: Test business logic without infrastructure
- Flexibility: Swap MongoDB for PostgreSQL without changing business logic
- Reusability: Use same handlers for API, CLI, or background jobs
- Maintainability: Infrastructure changes don't affect domain
- Framework Independence: Business logic doesn't depend on FastAPI
- Business Focus: Domain logic is pure and clear
π Understanding Clean ArchitectureΒΆ
Before diving into code, it's helpful to understand the architectural principles that guide Neuroglia:
The Dependency RuleΒΆ
graph TD
A[π API Layer<br/>Controllers, DTOs] --> B[πΌ Application Layer<br/>Commands, Queries, Handlers]
B --> C[ποΈ Domain Layer<br/>Entities, Business Rules]
D[π Integration Layer<br/>Repositories, External APIs] --> C
style C fill:#e1f5fe
style B fill:#f3e5f5
style A fill:#e8f5e8
style D fill:#fff3e0
Key principle: Inner layers never depend on outer layers. This enables:
- Testability - Easy to mock external dependencies
- Flexibility - Swap implementations without affecting business logic
- Maintainability - Changes in infrastructure don't break business rules
- Domain Focus - Business logic stays pure and framework-agnostic
CQRS in PracticeΒΆ
graph LR
A[Client Request] --> B{Command or Query?}
B -->|Write Operation| C[Command Handler]
B -->|Read Operation| D[Query Handler]
C --> E[Domain Logic]
E --> F[Repository]
D --> G[Read Model]
style C fill:#ffcdd2
style D fill:#c8e6c9
style E fill:#e1f5fe
Commands (Write): Create, Update, Delete operations that change system state Queries (Read): Retrieve operations that return data without side effects
This separation enables:
- Performance Optimization - Different models for reads vs writes
- Scalability - Scale read and write operations independently
- Clarity - Clear intent whether operation changes state
- Event Sourcing - Natural fit for event-driven architectures
π― OverviewΒΆ
Clean Architecture organizes code into four distinct layers, with the Mario's Pizzeria system serving as our primary example of how this pattern enables scalable, maintainable applications.
C4Container
title Clean Architecture - Mario's Pizzeria System
Container_Boundary(api, "π API Layer") {
Container(orders_controller, "Orders Controller", "FastAPI", "REST endpoints for pizza orders")
Container(menu_controller, "Menu Controller", "FastAPI", "Menu management and retrieval")
Container(kitchen_controller, "Kitchen Controller", "FastAPI", "Kitchen workflow management")
}
Container_Boundary(app, "πΌ Application Layer") {
Container(mediator, "Mediator", "CQRS", "Command/Query routing")
Container(handlers, "Command/Query Handlers", "Business Logic", "Order processing, menu queries")
Container(pipeline, "Pipeline Behaviors", "Cross-cutting", "Validation, logging, caching")
}
Container_Boundary(domain, "ποΈ Domain Layer") {
Container(entities, "Pizza Entities", "Domain Models", "Order, Pizza, Customer entities")
Container(events, "Domain Events", "Business Events", "OrderPlaced, PizzaReady events")
Container(rules, "Business Rules", "Domain Logic", "Pricing, validation rules")
}
Container_Boundary(integration, "π Integration Layer") {
Container(repos, "Repositories", "Data Access", "Order, Menu data persistence")
Container(external, "External Services", "Third-party", "Payment, SMS notifications")
Container(storage, "Data Storage", "Persistence", "MongoDB, File System")
}
Rel(orders_controller, mediator, "sends commands/queries")
Rel(menu_controller, mediator, "sends queries")
Rel(kitchen_controller, mediator, "sends commands")
Rel(mediator, handlers, "routes to")
Rel(handlers, entities, "uses")
Rel(handlers, events, "publishes")
Rel(handlers, repos, "persists via")
Rel(repos, storage, "stores in")
Rel(handlers, external, "integrates with")
β BenefitsΒΆ
1. TestabilityΒΆ
Each layer can be tested independently using mocks and stubs:
# Testing Order Handler without database dependencies
class TestPlaceOrderHandler:
def setup_method(self):
self.mock_repository = Mock(spec=OrderRepository)
self.mock_payment = Mock(spec=PaymentService)
self.handler = PlaceOrderHandler(self.mock_repository, self.mock_payment)
async def test_place_order_success(self):
# Arrange
command = PlaceOrderCommand(customer_id="123", pizzas=["margherita"])
# Act
result = await self.handler.handle_async(command)
# Assert
assert result.is_success
self.mock_repository.save_async.assert_called_once()
2. IndependenceΒΆ
Business logic in the domain layer is completely independent of frameworks, databases, and external services.
3. MaintainabilityΒΆ
Changes to external systems (databases, APIs) don't affect business logic.
π Data FlowΒΆ
The pizza ordering workflow demonstrates clean architecture data flow:
sequenceDiagram
participant Customer
participant API as OrdersController
participant Med as Mediator
participant Handler as PlaceOrderHandler
participant Domain as Order Entity
participant Repo as OrderRepository
participant DB as MongoDB
Customer->>+API: POST /orders (pizza order)
Note over API: π API Layer - HTTP endpoint
API->>+Med: Execute PlaceOrderCommand
Note over Med: πΌ Application Layer - CQRS routing
Med->>+Handler: Handle command
Note over Handler: πΌ Application Layer - Business workflow
Handler->>+Domain: Create Order entity
Note over Domain: ποΈ Domain Layer - Business rules
Domain-->>-Handler: Order with domain events
Handler->>+Repo: Save order
Note over Repo: π Integration Layer - Data access
Repo->>+DB: Insert document
Note over DB: π Integration Layer - Persistence
DB-->>-Repo: Success
Repo-->>-Handler: Order saved
Handler-->>-Med: OrderDto result
Med-->>-API: Success response
API-->>-Customer: 201 Created + OrderDto
π― Use CasesΒΆ
Clean Architecture is ideal for:
- Complex Business Logic: When domain rules are intricate (pricing, promotions, kitchen workflows)
- Multiple Interfaces: Supporting web APIs, mobile apps, and admin panels
- Long-term Maintenance: Systems that need to evolve over time
- Team Collaboration: Clear boundaries enable parallel development
π Implementation in Mario's PizzeriaΒΆ
Domain Layer (Core Business)ΒΆ
# domain/entities/order.py
class Order(Entity):
def __init__(self, customer_id: str, items: List[OrderItem]):
super().__init__()
self.customer_id = customer_id
self.items = items
self.status = OrderStatus.PENDING
self.total = self._calculate_total()
# Domain event for business workflow
self.raise_event(OrderPlacedEvent(
order_id=self.id,
customer_id=customer_id,
total=self.total
))
def _calculate_total(self) -> Decimal:
"""Business rule: Calculate order total with tax"""
subtotal = sum(item.price for item in self.items)
tax = subtotal * Decimal('0.08') # 8% tax
return subtotal + tax
Application Layer (Use Cases)ΒΆ
# application/handlers/place_order_handler.py
class PlaceOrderHandler(CommandHandler[PlaceOrderCommand, OperationResult[OrderDto]]):
def __init__(self,
order_repository: OrderRepository,
payment_service: PaymentService,
mapper: Mapper):
self._repository = order_repository
self._payment = payment_service
self._mapper = mapper
async def handle_async(self, command: PlaceOrderCommand) -> OperationResult[OrderDto]:
# Create domain entity (business logic)
order = Order(command.customer_id, command.items)
# Process payment (external integration)
payment_result = await self._payment.process_async(order.total)
if not payment_result.success:
return self.bad_request("Payment failed")
# Persist order (data access)
await self._repository.save_async(order)
# Return result
dto = self._mapper.map(order, OrderDto)
return self.created(dto)
API Layer (Interface)ΒΆ
# api/controllers/orders_controller.py
class OrdersController(ControllerBase):
@post("/", response_model=OrderDto, status_code=201)
async def place_order(self, request: PlaceOrderRequest) -> OrderDto:
command = self.mapper.map(request, PlaceOrderCommand)
result = await self.mediator.execute_async(command)
return self.process(result)
Integration Layer (External Concerns)ΒΆ
# integration/repositories/mongo_order_repository.py
class MongoOrderRepository(Repository[Order, str]):
def __init__(self, collection: Collection):
self._collection = collection
async def save_async(self, order: Order) -> None:
document = {
"_id": order.id,
"customer_id": order.customer_id,
"items": [{"name": item.name, "price": float(item.price)}
for item in order.items],
"total": float(order.total),
"status": order.status.value
}
await self._collection.insert_one(document)
π§ͺ Testing Clean ArchitectureΒΆ
Unit Testing Domain LayerΒΆ
import pytest
from decimal import Decimal
class TestOrderEntity:
def test_order_calculates_total_with_tax(self):
# Arrange
items = [
OrderItem(pizza_name="Margherita", price=Decimal("12.99"), quantity=2),
OrderItem(pizza_name="Pepperoni", price=Decimal("14.99"), quantity=1)
]
# Act
order = Order(customer_id="cust_123", items=items)
# Assert
expected_subtotal = Decimal("40.97") # 12.99*2 + 14.99
expected_tax = expected_subtotal * Decimal("0.08")
expected_total = expected_subtotal + expected_tax
assert order.total == expected_total
def test_order_raises_domain_event(self):
# Arrange
items = [OrderItem(pizza_name="Margherita", price=Decimal("12.99"), quantity=1)]
# Act
order = Order(customer_id="cust_123", items=items)
# Assert
events = order.get_uncommitted_events()
assert len(events) == 1
assert isinstance(events[0], OrderPlacedEvent)
assert events[0].order_id == order.id
Unit Testing Application LayerΒΆ
@pytest.mark.asyncio
async def test_place_order_handler_success():
# Arrange
mock_repository = AsyncMock(spec=IOrderRepository)
mock_payment = AsyncMock(spec=IPaymentService)
mock_payment.process_async.return_value = PaymentResult(success=True)
handler = PlaceOrderHandler(mock_repository, mock_payment, Mock())
command = PlaceOrderCommand(
customer_id="cust_123",
items=[OrderItemDto(pizza_name="Margherita", price="12.99", quantity=1)],
payment_method="card"
)
# Act
result = await handler.handle_async(command)
# Assert
assert result.is_success
mock_payment.process_async.assert_called_once()
mock_repository.save_async.assert_called_once()
@pytest.mark.asyncio
async def test_place_order_handler_payment_failure():
# Arrange
mock_repository = AsyncMock(spec=IOrderRepository)
mock_payment = AsyncMock(spec=IPaymentService)
mock_payment.process_async.return_value = PaymentResult(
success=False,
error="Card declined"
)
handler = PlaceOrderHandler(mock_repository, mock_payment, Mock())
command = PlaceOrderCommand(customer_id="cust_123", items=[], payment_method="card")
# Act
result = await handler.handle_async(command)
# Assert
assert not result.is_success
assert "Payment failed" in result.error_message
mock_repository.save_async.assert_not_called()
Integration TestingΒΆ
@pytest.mark.integration
@pytest.mark.asyncio
async def test_complete_order_workflow():
# Arrange - use test app with real mediator and in-memory repositories
app = create_test_app()
client = TestClient(app)
order_data = {
"customer_id": "test_customer",
"items": [
{"pizza_name": "Margherita", "price": "12.99", "quantity": 2}
],
"payment_method": "card"
}
# Act
response = client.post("/api/orders", json=order_data)
# Assert
assert response.status_code == 201
result = response.json()
assert "order_id" in result
assert result["total"] == "27.95" # (12.99 * 2) * 1.08
β οΈ Common MistakesΒΆ
1. Layer Violations (Breaking Dependency Rule)ΒΆ
# β Wrong - Domain layer depends on infrastructure
from pymongo import Collection # β Infrastructure import in domain
class Order(Entity):
def __init__(self, customer_id: str, collection: Collection):
# β Domain entity depends on MongoDB
self.collection = collection
async def save(self):
# β Domain entity performing data access
await self.collection.insert_one(self.__dict__)
# β
Correct - Domain layer has no infrastructure dependencies
class Order(Entity):
def __init__(self, customer_id: str, items: List[OrderItem]):
# β
Pure business logic only
self.customer_id = customer_id
self.items = items
self.total = self._calculate_total()
# β
Repository handles persistence (integration layer)
2. Business Logic in ControllersΒΆ
# β Wrong - Business logic in API layer
class OrdersController(ControllerBase):
@post("/orders")
async def place_order(self, request: dict):
# β Tax calculation in controller
subtotal = sum(item["price"] for item in request["items"])
tax = subtotal * 0.08
total = subtotal + tax
# β Validation in controller
if total > 1000:
return {"error": "Order too large"}
order_doc = {"total": total, "items": request["items"]}
await self._db.orders.insert_one(order_doc)
return order_doc
# β
Correct - Thin controller delegates to application layer
class OrdersController(ControllerBase):
@post("/orders", response_model=OrderDto)
async def place_order(self, request: PlaceOrderRequest):
# β
Map and delegate
command = self.mapper.map(request, PlaceOrderCommand)
result = await self.mediator.execute_async(command)
return self.process(result)
3. Direct Repository Dependencies in ControllersΒΆ
# β Wrong - Controller directly uses repository
class OrdersController:
def __init__(self, order_repository: IOrderRepository):
# β Controller depends on repository
self._repository = order_repository
@post("/orders")
async def place_order(self, request: dict):
order = Order(**request)
# β Controller calling repository directly
await self._repository.save_async(order)
return order
# β
Correct - Controller uses mediator
class OrdersController(ControllerBase):
# β
Only depends on base class (provides mediator)
@post("/orders", response_model=OrderDto)
async def place_order(self, request: PlaceOrderRequest):
# β
Uses mediator for all operations
command = self.mapper.map(request, PlaceOrderCommand)
result = await self.mediator.execute_async(command)
return self.process(result)
4. Returning Domain Entities from APIΒΆ
# β Wrong - Exposing domain entities to API
class OrdersController(ControllerBase):
@get("/orders/{order_id}")
async def get_order(self, order_id: str) -> Order: # β Returns entity
query = GetOrderQuery(order_id=order_id)
order = await self.mediator.execute_async(query)
return order # β Exposing domain entity
# β
Correct - Return DTOs
class OrdersController(ControllerBase):
@get("/orders/{order_id}", response_model=OrderDto) # β
Returns DTO
async def get_order(self, order_id: str) -> OrderDto:
query = GetOrderQuery(order_id=order_id)
result = await self.mediator.execute_async(query)
return self.process(result) # β
Returns mapped DTO
5. Mixing Infrastructure Code in Application LayerΒΆ
# β Wrong - Application layer with infrastructure details
class PlaceOrderHandler(CommandHandler):
async def handle_async(self, command: PlaceOrderCommand):
# β Direct MongoDB access in handler
from pymongo import MongoClient
client = MongoClient("mongodb://localhost")
db = client.pizzeria
order_doc = {"customer_id": command.customer_id}
await db.orders.insert_one(order_doc)
# β
Correct - Application layer uses abstractions
class PlaceOrderHandler(CommandHandler):
def __init__(self, repository: IOrderRepository):
# β
Depends on interface
self._repository = repository
async def handle_async(self, command: PlaceOrderCommand):
order = Order(command.customer_id, command.items)
# β
Uses repository abstraction
await self._repository.save_async(order)
6. Anemic Domain ModelsΒΆ
# β Wrong - Domain entity with no behavior
class Order:
def __init__(self):
self.customer_id = None
self.items = []
self.total = 0
# β Just a data bag, no business logic
# Business logic scattered in handlers
class PlaceOrderHandler:
async def handle_async(self, command: PlaceOrderCommand):
order = Order()
order.customer_id = command.customer_id
order.items = command.items
# β Calculating total in handler
order.total = sum(item.price for item in order.items) * 1.08
# β
Correct - Rich domain model with behavior
class Order(Entity):
def __init__(self, customer_id: str, items: List[OrderItem]):
# β
Business logic in entity
self.customer_id = customer_id
self.items = items
self.total = self._calculate_total()
self.raise_event(OrderPlacedEvent(...))
def _calculate_total(self) -> Decimal:
# β
Business rule encapsulated
subtotal = sum(item.price * item.quantity for item in self.items)
return subtotal * Decimal("1.08")
def apply_discount(self, percentage: Decimal):
# β
Business behavior on entity
if percentage > Decimal("0.5"):
raise ValueError("Discount cannot exceed 50%")
self.total = self.total * (Decimal("1") - percentage)
π« When NOT to UseΒΆ
1. Simple CRUD ApplicationsΒΆ
For basic applications with minimal business logic:
# Clean architecture is overkill for simple CRUD
@app.get("/pizzas")
async def get_pizzas(db: Database):
return await db.pizzas.find().to_list(None)
@app.post("/pizzas")
async def create_pizza(pizza: PizzaDto, db: Database):
result = await db.pizzas.insert_one(pizza.dict())
return {"id": str(result.inserted_id)}
2. Prototypes and Proof of ConceptsΒΆ
When rapidly testing ideas without long-term maintenance needs:
# Quick prototype - simple FastAPI endpoints sufficient
@app.post("/orders")
async def place_order(order_data: dict, db: Database):
# Direct implementation without layers
result = await db.orders.insert_one(order_data)
return {"order_id": str(result.inserted_id)}
3. Single-Purpose ScriptsΒΆ
For one-off data migration or batch processing scripts:
# Simple script doesn't need architecture layers
import pymongo
client = pymongo.MongoClient("mongodb://localhost")
db = client.pizzeria
# Direct operations
for order in db.orders.find({"status": "pending"}):
db.orders.update_one({"_id": order["_id"]}, {"$set": {"status": "completed"}})
4. Very Small Teams Without Architecture ExperienceΒΆ
When team lacks experience with layered architecture:
# Simple service pattern may be better
class OrderService:
def __init__(self, db: Database):
self.db = db
async def create_order(self, order_data: dict):
return await self.db.orders.insert_one(order_data)
π Key TakeawaysΒΆ
- Dependency Rule: Dependencies flow inward - outer layers depend on inner layers
- Four Layers: API β Application β Domain β Integration
- Domain Independence: Business logic has no framework or infrastructure dependencies
- Testability: Test each layer independently with mocks
- Abstractions: Application layer depends on interfaces, not implementations
- DTOs at Boundaries: API layer uses DTOs, not domain entities
- Rich Domain Models: Entities contain business logic, not just data
- Single Responsibility: Each layer has clear, focused responsibilities
- Framework Independence: Business logic doesn't depend on FastAPI, Django, etc.
- Long-Term Maintainability: Architecture supports evolution and scaling
π Related PatternsΒΆ
- CQRS Pattern - Separates commands and queries within the application layer
- Event-Driven Architecture - Uses domain events for decoupled communication
- Repository Pattern - Abstracts data access in the integration layer
- Domain-Driven Design - Rich domain models with business behavior
- Dependency Injection - Wires abstractions to implementations
This pattern guide demonstrates Clean Architecture using Mario's Pizzeria as a practical example. The four-layer approach shown here scales from simple applications to complex enterprise systems. ποΈ