Software architecture patterns provide proven structural solutions for organizing applications to achieve scalability, maintainability, and testability. Understanding these patterns enables architects to make informed decisions about system design based on specific business requirements and technical constraints. github clustox.com
Layered Architecture
Layered Architecture organizes an application into horizontal layers where each layer has a distinct responsibility and communicates only with adjacent layers. dev.to
Core Layers
The traditional layered architecture consists of four primary layers:
- Presentation Layer handles user interface interactions and browser communications, serving as the entry point for user requests
- Business Layer executes business logic and rules, processing requests from the presentation layer.
- Persistence Layer manages data persistence operations and coordinates with the database layer.
- Database Layer stores and retrieves data from physical storage systems.
Architecture Diagram
Python Implementation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
| # Database Layer
class Database:
def __init__(self):
self.users = {}
def save_user(self, user_id, user_data):
self.users[user_id] = user_data
return True
def get_user(self, user_id):
return self.users.get(user_id)
# Persistence Layer
class UserRepository:
def __init__(self, database):
self.db = database
def create_user(self, user_id, username, email):
user_data = {'username': username, 'email': email}
return self.db.save_user(user_id, user_data)
def find_user(self, user_id):
return self.db.get_user(user_id)
# Business Layer
class UserService:
def __init__(self, repository):
self.repository = repository
def register_user(self, user_id, username, email):
# Business validation
if not email or '@' not in email:
raise ValueError("Invalid email")
if len(username) < 3:
raise ValueError("Username too short")
return self.repository.create_user(user_id, username, email)
def get_user_profile(self, user_id):
user = self.repository.find_user(user_id)
if not user:
raise ValueError("User not found")
return user
# Presentation Layer
class UserController:
def __init__(self, service):
self.service = service
def handle_registration(self, request_data):
try:
user = self.service.register_user(
request_data['id'],
request_data['username'],
request_data['email']
)
return {'status': 'success', 'data': user}
except ValueError as e:
return {'status': 'error', 'message': str(e)}
# Application Assembly
db = Database()
repo = UserRepository(db)
service = UserService(repo)
controller = UserController(service)
# Usage
response = controller.handle_registration({
'id': '123',
'username': 'johndoe',
'email': 'john@example.com'
})
print(response)
|
Use Cases
Layered architecture excels in scenarios requiring quick application development with traditional IT infrastructure. This pattern suits applications needing clear separation of concerns with long-term maintainability, particularly when budget or time constraints exist. Applications with isolated changes confined to specific layers benefit from this approach, as UI redesigns or database migrations impact only their respective layers
Best Practices
Structure applications with clear layer boundaries to enable straightforward unit testing. Align team structure with technical expertise (front-end, back-end, database) to leverage Conway’s Law for smooth collaboration. Use layered architecture as a starting point when the optimal architecture remains unclear, as it provides a versatile, well-understood foundation.
When to Use
Apply layered architecture for general-purpose applications with well-established methods and minimal complexity. This pattern works effectively when most changes remain confined to specific layers and teams are organized by technical specialization.
When to Avoid
Avoid layered architecture for systems requiring high scalability, fault tolerance, or exceptional performance, as the monolithic nature creates scaling challenges. Applications with complex features requiring frequent cross-layer changes become cumbersome with this pattern. Domain-driven designs where changes span technical boundaries rather than remaining layer-confined are poorly suited to layered architecture.
Hexagonal Architecture (Ports & Adapters)
Hexagonal Architecture, proposed by Dr. Alistair Cockburn in 2005, creates loosely coupled systems where application components can be tested independently without dependencies on data stores or user interfaces.
aws dev.to
Core Concepts
The architecture uses ports as interfaces defined by the application’s core (the hexagon) and adapters that implement these interfaces to connect external systems. This pattern isolates business logic from infrastructure code, enabling technology stack changes with minimal impact to business logic. angular.love
Ports represent abstract interfaces that define requirements without implementation details. Adapters provide concrete implementations of ports, translating technical exchanges with external components
Architecture Diagram
Python Implementation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
| from abc import ABC, abstractmethod
from typing import List, Optional
# Domain Model (Core)
class Order:
def __init__(self, order_id: str, customer_id: str, items: List[str]):
self.order_id = order_id
self.customer_id = customer_id
self.items = items
self.status = 'pending'
def confirm(self):
if not self.items:
raise ValueError("Cannot confirm empty order")
self.status = 'confirmed'
# Output Port (Interface for Infrastructure)
class OrderRepository(ABC):
@abstractmethod
def save(self, order: Order) -> bool:
pass
@abstractmethod
def find_by_id(self, order_id: str) -> Optional[Order]:
pass
class NotificationService(ABC):
@abstractmethod
def notify_customer(self, customer_id: str, message: str) -> bool:
pass
# Input Port (Use Case)
class CreateOrderUseCase:
def __init__(self, repository: OrderRepository,
notification: NotificationService):
self.repository = repository
self.notification = notification
def execute(self, order_id: str, customer_id: str,
items: List[str]) -> Order:
# Business logic in the core
order = Order(order_id, customer_id, items)
order.confirm()
# Use output ports
self.repository.save(order)
self.notification.notify_customer(
customer_id,
f"Order {order_id} confirmed"
)
return order
# Adapters (Infrastructure Implementation)
class PostgresOrderRepository(OrderRepository):
def __init__(self):
self.storage = {}
def save(self, order: Order) -> bool:
self.storage[order.order_id] = order
print(f"Saved to PostgreSQL: {order.order_id}")
return True
def find_by_id(self, order_id: str) -> Optional[Order]:
return self.storage.get(order_id)
class EmailNotificationService(NotificationService):
def notify_customer(self, customer_id: str, message: str) -> bool:
print(f"Email sent to {customer_id}: {message}")
return True
class SMSNotificationService(NotificationService):
def notify_customer(self, customer_id: str, message: str) -> bool:
print(f"SMS sent to {customer_id}: {message}")
return True
# REST API Adapter (Input)
class OrderController:
def __init__(self, create_order_use_case: CreateOrderUseCase):
self.create_order = create_order_use_case
def handle_create_order(self, request_data: dict) -> dict:
try:
order = self.create_order.execute(
request_data['order_id'],
request_data['customer_id'],
request_data['items']
)
return {
'status': 'success',
'order_id': order.order_id,
'order_status': order.status
}
except Exception as e:
return {'status': 'error', 'message': str(e)}
# Application Assembly (Dependency Injection)
repository = PostgresOrderRepository()
notification = EmailNotificationService() # Can swap to SMSNotificationService
use_case = CreateOrderUseCase(repository, notification)
controller = OrderController(use_case)
# Usage
result = controller.handle_create_order({
'order_id': 'ORD-001',
'customer_id': 'CUST-123',
'items': ['item1', 'item2']
})
print(result)
|
Use Cases
Hexagonal architecture excels for AWS Lambda functions requiring integration with external services through loosely coupled business logic. This pattern enables designing systems by purpose rather than technology, resulting in easily exchangeable components like databases, UX, and services. Applications requiring technology stack changes over time benefit from this architecture’s prevention of technology lock-in.
Best Practices
Define ports as abstract interfaces without implementation details to maintain core business logic independence. Implement adapters outside the hexagon to translate between external systems and the core domain. Use dependency injection to wire adapters to ports, enabling easy swapping of infrastructure components for testing or production.
When to Use
Apply hexagonal architecture when business logic requires isolation from infrastructure concerns. This pattern suits applications needing independent testing of components without database or UI dependencies. Systems anticipating future technology changes or requiring multiple interface types (REST API, GraphQL, CLI) benefit from ports and adapters.
CQRS & Event Sourcing
CQRS (Command Query Responsibility Segregation) separates read and write operations into distinct models, while Event Sourcing stores state changes as a chronological series of events rather than current state. linkedin microsoft
Core Concepts
CQRS splits the system into Command Model for state-changing operations and Query Model for read operations. Event Sourcing persists events as the single source of truth in an append-only event store, enabling state reconstruction by replaying events.
The Command side handles writes and generates events, while the Query side builds projections from events for optimized reads. This separation allows independent scaling and tuning of read versus write workloads.
Architecture Diagram
Python Implementation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
| from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict
from enum import Enum
# Domain Events
@dataclass
class Event:
event_id: str
aggregate_id: str
event_type: str
timestamp: datetime
data: dict
@dataclass
class ItemAddedToCart(Event):
pass
@dataclass
class ItemRemovedFromCart(Event):
pass
@dataclass
class OrderPlaced(Event):
pass
# Event Store
class EventStore:
def __init__(self):
self.events: List[Event] = []
def append(self, event: Event):
self.events.append(event)
print(f"Event stored: {event.event_type}")
def get_events(self, aggregate_id: str) -> List[Event]:
return [e for e in self.events if e.aggregate_id == aggregate_id]
# Command Side - Aggregate
class ShoppingCart:
def __init__(self, cart_id: str):
self.cart_id = cart_id
self.items: Dict[str, int] = {}
self.uncommitted_events: List[Event] = []
def add_item(self, item_id: str, quantity: int):
if quantity <= 0:
raise ValueError("Quantity must be positive")
event = ItemAddedToCart(
event_id=f"evt-{len(self.uncommitted_events)}",
aggregate_id=self.cart_id,
event_type="ItemAddedToCart",
timestamp=datetime.now(),
data={'item_id': item_id, 'quantity': quantity}
)
self.apply_event(event)
self.uncommitted_events.append(event)
def remove_item(self, item_id: str):
if item_id not in self.items:
raise ValueError("Item not in cart")
event = ItemRemovedFromCart(
event_id=f"evt-{len(self.uncommitted_events)}",
aggregate_id=self.cart_id,
event_type="ItemRemovedFromCart",
timestamp=datetime.now(),
data={'item_id': item_id}
)
self.apply_event(event)
self.uncommitted_events.append(event)
def apply_event(self, event: Event):
if event.event_type == "ItemAddedToCart":
item_id = event.data['item_id']
quantity = event.data['quantity']
self.items[item_id] = self.items.get(item_id, 0) + quantity
elif event.event_type == "ItemRemovedFromCart":
item_id = event.data['item_id']
if item_id in self.items:
del self.items[item_id]
def get_uncommitted_events(self) -> List[Event]:
return self.uncommitted_events
def mark_events_committed(self):
self.uncommitted_events = []
# Command Handlers
class AddItemCommand:
def __init__(self, cart_id: str, item_id: str, quantity: int):
self.cart_id = cart_id
self.item_id = item_id
self.quantity = quantity
class CommandHandler:
def __init__(self, event_store: EventStore, event_bus):
self.event_store = event_store
self.event_bus = event_bus
self.carts: Dict[str, ShoppingCart] = {}
def handle_add_item(self, command: AddItemCommand):
# Get or create aggregate
if command.cart_id not in self.carts:
self.carts[command.cart_id] = ShoppingCart(command.cart_id)
cart = self.carts[command.cart_id]
# Execute business logic
cart.add_item(command.item_id, command.quantity)
# Persist events
for event in cart.get_uncommitted_events():
self.event_store.append(event)
self.event_bus.publish(event)
cart.mark_events_committed()
# Query Side - Projections
class CartProjection:
def __init__(self):
self.carts: Dict[str, Dict] = {}
def handle_event(self, event: Event):
cart_id = event.aggregate_id
if cart_id not in self.carts:
self.carts[cart_id] = {'items': {}, 'total_items': 0}
if event.event_type == "ItemAddedToCart":
item_id = event.data['item_id']
quantity = event.data['quantity']
current = self.carts[cart_id]['items'].get(item_id, 0)
self.carts[cart_id]['items'][item_id] = current + quantity
self.carts[cart_id]['total_items'] += quantity
print(f"Projection updated: Cart {cart_id} has {self.carts[cart_id]['total_items']} items")
elif event.event_type == "ItemRemovedFromCart":
item_id = event.data['item_id']
if item_id in self.carts[cart_id]['items']:
quantity = self.carts[cart_id]['items'][item_id]
del self.carts[cart_id]['items'][item_id]
self.carts[cart_id]['total_items'] -= quantity
def get_cart(self, cart_id: str) -> Dict:
return self.carts.get(cart_id, {'items': {}, 'total_items': 0})
# Query Handlers
class GetCartQuery:
def __init__(self, cart_id: str):
self.cart_id = cart_id
class QueryHandler:
def __init__(self, projection: CartProjection):
self.projection = projection
def handle_get_cart(self, query: GetCartQuery) -> Dict:
return self.projection.get_cart(query.cart_id)
# Event Bus (Simple Implementation)
class EventBus:
def __init__(self):
self.subscribers = []
def subscribe(self, handler):
self.subscribers.append(handler)
def publish(self, event: Event):
for subscriber in self.subscribers:
subscriber.handle_event(event)
# Application Assembly
event_store = EventStore()
event_bus = EventBus()
projection = CartProjection()
event_bus.subscribe(projection)
command_handler = CommandHandler(event_store, event_bus)
query_handler = QueryHandler(projection)
# Usage - Command Side
add_command = AddItemCommand('cart-001', 'item-123', 2)
command_handler.handle_add_item(add_command)
add_command2 = AddItemCommand('cart-001', 'item-456', 1)
command_handler.handle_add_item(add_command2)
# Usage - Query Side
get_query = GetCartQuery('cart-001')
cart_data = query_handler.handle_get_cart(get_query)
print(f"Cart contents: {cart_data}")
|
Use Cases
CQRS with Event Sourcing excels for systems requiring full audit trails and state reconstruction capabilities. E-commerce platforms benefit from independent scaling of read-heavy product browsing versus write-heavy order processing. Financial systems requiring complete transaction history and point-in-time state reconstruction leverage Event Sourcing’s append-only event log.
When to Use
Apply CQRS when read and write workloads differ significantly in scale or requirements. Systems requiring temporal queries, audit trails, or state reconstruction at specific points in time benefit from Event Sourcing. Applications needing independent scaling of read versus write operations leverage CQRS’s separation.
When to Avoid
Avoid CQRS for simple CRUD applications where read and write operations have similar complexity, as the pattern adds unnecessary overhead. Systems requiring strong consistency between reads and writes struggle with eventual consistency inherent in CQRS. Small applications without complex domain logic or significant scale requirements find CQRS’s complexity unjustified.