Architecture Patterns
System-level patterns for building scalable, maintainable applications
From Design to Architecture
While design patterns solve code-level problems, architecture patterns address system-level concerns: how to structure entire applications, how services communicate, how to achieve scalability and maintainability. These patterns represent decades of experience building large-scale systems at companies like Netflix, Amazon, and Google. Understanding architecture patterns is what separates senior engineers who write good code from architects who design good systems. Each pattern represents trade-offs, there's no "best" architecture, only the right architecture for your specific context.
System Architecture Patterns
🏛️ Structural Patterns
🔄 Interaction Patterns
💾 Data Patterns
🎯 Domain Patterns
Monolithic Architecture
Intent: Build the entire application as a single, unified codebase and deployment unit. All components (UI, business logic, data access) run in the same process and share the same database.
Structure
# Typical monolithic application structure ecommerce/ ├── app.py # Main application ├── models/ # Domain models │ ├── user.py │ ├── product.py │ └── order.py ├── views/ # UI layer │ ├── user_views.py │ ├── product_views.py │ └── order_views.py ├── services/ # Business logic │ ├── auth_service.py │ ├── inventory_service.py │ └── payment_service.py ├── repositories/ # Data access │ ├── user_repository.py │ └── order_repository.py └── database.py # Single shared database # Everything runs in one process, shares one database # Deployed as a single unit
Example Implementation
# Monolithic e-commerce application
class UserService:
def __init__(self, db):
self.db = db
def create_user(self, email, password):
# Direct database access
user = self.db.users.create(email=email, password=hash(password))
return user
class OrderService:
def __init__(self, db, inventory_service, payment_service):
self.db = db
self.inventory = inventory_service
self.payment = payment_service
def create_order(self, user_id, items):
# All services in same process - direct method calls
if not self.inventory.check_availability(items):
raise ValueError("Items not available")
total = sum(item.price * item.quantity for item in items)
if not self.payment.charge(user_id, total):
raise ValueError("Payment failed")
# Single database transaction
with self.db.transaction():
order = self.db.orders.create(user_id=user_id, total=total)
for item in items:
self.db.order_items.create(order_id=order.id, item=item)
self.inventory.decrease_stock(item.id, item.quantity)
return order
class InventoryService:
def __init__(self, db):
self.db = db
def check_availability(self, items):
for item in items:
product = self.db.products.get(item.id)
if product.stock < item.quantity:
return False
return True
def decrease_stock(self, product_id, quantity):
self.db.products.update(product_id, stock=stock - quantity)
# Everything instantiated together
db = Database()
inventory = InventoryService(db)
payment = PaymentService(db)
order_service = OrderService(db, inventory, payment)
user_service = UserService(db)Pros & Cons
✅ Advantages
- Simple to develop and test
- Easy deployment
- Strong consistency (ACID)
- No network overhead
- Easier debugging
❌ Disadvantages
- Tight coupling
- Difficult to scale specific parts
- Long deployment cycles
- Technology stack locked
- Large codebase complexity
When to Use Monoliths: Small to medium applications, startups (MVP), teams without distributed systems expertise, applications with simple workflows.
Microservices Architecture
Intent: Structure the application as a collection of loosely coupled, independently deployable services. Each service owns its data and communicates via well-defined APIs (typically REST or gRPC).
Core Principles
# Microservices architecture - separate services
# User Service (users.example.com)
class UserService:
def __init__(self):
self.db = UserDatabase() # Own database
@app.post("/users")
def create_user(self, email, password):
user = self.db.create(email=email, password=hash(password))
# Publish event for other services
event_bus.publish("user.created", {"user_id": user.id, "email": email})
return user
@app.get("/users/{user_id}")
def get_user(self, user_id):
return self.db.get(user_id)
# Inventory Service (inventory.example.com)
class InventoryService:
def __init__(self):
self.db = InventoryDatabase() # Own database
@app.get("/inventory/{product_id}")
def check_availability(self, product_id):
product = self.db.get(product_id)
return {"available": product.stock > 0, "stock": product.stock}
@app.post("/inventory/{product_id}/reserve")
def reserve_stock(self, product_id, quantity):
# Reserve stock temporarily for order
return self.db.reserve(product_id, quantity)
# Listen to order events
@event_bus.subscribe("order.completed")
def on_order_completed(self, event):
self.db.decrease_stock(event["product_id"], event["quantity"])
# Order Service (orders.example.com)
class OrderService:
def __init__(self):
self.db = OrderDatabase() # Own database
self.user_client = HTTPClient("users.example.com")
self.inventory_client = HTTPClient("inventory.example.com")
self.payment_client = HTTPClient("payments.example.com")
@app.post("/orders")
async def create_order(self, user_id, items):
# Call other services via HTTP/gRPC
user = await self.user_client.get(f"/users/{user_id}")
if not user:
raise ValueError("User not found")
# Check inventory across service boundary
for item in items:
availability = await self.inventory_client.get(
f"/inventory/{item.id}"
)
if not availability["available"]:
raise ValueError(f"Item {item.id} not available")
# Calculate total
total = sum(item.price * item.quantity for item in items)
# Process payment via payment service
payment = await self.payment_client.post("/payments", {
"user_id": user_id,
"amount": total
})
if payment["status"] != "success":
raise ValueError("Payment failed")
# Create order in own database
order = self.db.create(user_id=user_id, total=total)
# Publish event for other services
event_bus.publish("order.completed", {
"order_id": order.id,
"items": items
})
return order
# Each service:
# - Runs independently
# - Has its own database
# - Communicates via APIs/events
# - Can be deployed separately
# - Can use different tech stacksCommunication Patterns
# 1. Synchronous: REST/gRPC
class OrderService:
async def create_order(self, user_id, items):
# Synchronous HTTP call - waits for response
user = await http_client.get(f"http://user-service/users/{user_id}")
# Service dependency creates coupling
if not user:
raise ValueError("User service unavailable")
# 2. Asynchronous: Message Queue
class OrderService:
def create_order(self, user_id, items):
order = self.db.create(user_id=user_id, items=items)
# Publish event - fire and forget
message_queue.publish("order.created", {
"order_id": order.id,
"user_id": user_id,
"items": items
})
return order # Don't wait for other services
class InventoryService:
@message_queue.subscribe("order.created")
def handle_order_created(self, event):
# Process asynchronously when ready
for item in event["items"]:
self.decrease_stock(item["id"], item["quantity"])
# 3. Event-Driven: Event Bus
class PaymentService:
@event_bus.subscribe("order.created")
def process_payment(self, event):
result = self.charge(event["user_id"], event["total"])
if result.success:
event_bus.publish("payment.completed", {
"order_id": event["order_id"],
"transaction_id": result.id
})
else:
event_bus.publish("payment.failed", {
"order_id": event["order_id"]
})Pros & Cons
✅ Advantages
- Independent scaling
- Independent deployment
- Technology diversity
- Team autonomy
- Fault isolation
❌ Disadvantages
- Distributed system complexity
- Network latency overhead
- Eventual consistency
- Difficult testing/debugging
- Operational overhead
Reality Check: Microservices solve organizational problems (multiple teams) at the cost of technical complexity. Don't adopt microservices just because they're popular. Start with a monolith and extract services when you have a clear need.
Event-Driven Architecture (EDA)
Intent: Structure the system around the production, detection, and reaction to events. Components communicate by emitting and consuming events rather than direct calls. This creates highly decoupled, scalable systems.
Core Concepts
# Event: Immutable fact about something that happened
class Event:
def __init__(self, event_type, data, timestamp=None):
self.id = str(uuid.uuid4())
self.type = event_type
self.data = data
self.timestamp = timestamp or datetime.now()
# Event examples
user_created = Event("user.created", {
"user_id": "123",
"email": "user@example.com"
})
order_placed = Event("order.placed", {
"order_id": "456",
"user_id": "123",
"items": [{"product_id": "789", "quantity": 2}],
"total": 99.99
})
payment_completed = Event("payment.completed", {
"order_id": "456",
"transaction_id": "tx_123",
"amount": 99.99
})
# Event Bus: Central nervous system
class EventBus:
def __init__(self):
self._subscribers = defaultdict(list)
def publish(self, event: Event):
"""Publish event to all subscribers"""
print(f"Publishing event: {event.type}")
for handler in self._subscribers[event.type]:
try:
handler(event)
except Exception as e:
print(f"Error handling {event.type}: {e}")
def subscribe(self, event_type: str, handler: callable):
"""Subscribe to specific event type"""
self._subscribers[event_type].append(handler)
def unsubscribe(self, event_type: str, handler: callable):
"""Unsubscribe from event type"""
self._subscribers[event_type].remove(handler)
# Services consume and produce events
class UserService:
def __init__(self, event_bus: EventBus):
self.event_bus = event_bus
self.db = UserDatabase()
def create_user(self, email, password):
user = self.db.create(email=email, password=hash(password))
# Emit event instead of calling other services directly
self.event_bus.publish(Event("user.created", {
"user_id": user.id,
"email": user.email,
"created_at": user.created_at.isoformat()
}))
return user
class EmailService:
def __init__(self, event_bus: EventBus):
self.event_bus = event_bus
# Subscribe to events
self.event_bus.subscribe("user.created", self.send_welcome_email)
self.event_bus.subscribe("order.placed", self.send_order_confirmation)
def send_welcome_email(self, event: Event):
email = event.data["email"]
print(f"Sending welcome email to {email}")
# Email sending logic
def send_order_confirmation(self, event: Event):
order_id = event.data["order_id"]
print(f"Sending order confirmation for {order_id}")
class AnalyticsService:
def __init__(self, event_bus: EventBus):
self.event_bus = event_bus
# Subscribe to multiple events
self.event_bus.subscribe("user.created", self.track_user)
self.event_bus.subscribe("order.placed", self.track_order)
self.event_bus.subscribe("payment.completed", self.track_payment)
def track_user(self, event: Event):
# Analytics logic
print(f"Tracking user creation: {event.data['user_id']}")
def track_order(self, event: Event):
print(f"Tracking order: {event.data['order_id']}")
def track_payment(self, event: Event):
print(f"Tracking payment: {event.data['transaction_id']}")
# Wiring it all together
event_bus = EventBus()
user_service = UserService(event_bus)
email_service = EmailService(event_bus)
analytics_service = AnalyticsService(event_bus)
# Creating a user triggers multiple side effects automatically
user = user_service.create_user("alice@example.com", "password123")
# Events fire: email sent, analytics tracked, etc.Event Sourcing
# Event Sourcing: Store events as source of truth, not current state
class EventStore:
def __init__(self):
self.events = [] # In reality: durable storage
def append(self, event: Event):
"""Append event to immutable log"""
self.events.append(event)
def get_events(self, aggregate_id: str):
"""Get all events for an entity"""
return [e for e in self.events if e.data.get("aggregate_id") == aggregate_id]
class BankAccount:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0
self.version = 0
def apply_event(self, event: Event):
"""Rebuild state from events"""
if event.type == "account.opened":
self.balance = event.data["initial_balance"]
elif event.type == "money.deposited":
self.balance += event.data["amount"]
elif event.type == "money.withdrawn":
self.balance -= event.data["amount"]
self.version += 1
@staticmethod
def from_events(events):
"""Reconstruct account from event history"""
if not events:
return None
account = BankAccount(events[0].data["account_id"])
for event in events:
account.apply_event(event)
return account
class BankAccountService:
def __init__(self, event_store: EventStore):
self.event_store = event_store
def open_account(self, account_id, initial_balance):
event = Event("account.opened", {
"aggregate_id": account_id,
"account_id": account_id,
"initial_balance": initial_balance
})
self.event_store.append(event)
return account_id
def deposit(self, account_id, amount):
# Load current state from events
events = self.event_store.get_events(account_id)
account = BankAccount.from_events(events)
if not account:
raise ValueError("Account not found")
# Create new event
event = Event("money.deposited", {
"aggregate_id": account_id,
"amount": amount
})
self.event_store.append(event)
def withdraw(self, account_id, amount):
events = self.event_store.get_events(account_id)
account = BankAccount.from_events(events)
if account.balance < amount:
raise ValueError("Insufficient funds")
event = Event("money.withdrawn", {
"aggregate_id": account_id,
"amount": amount
})
self.event_store.append(event)
def get_balance(self, account_id):
events = self.event_store.get_events(account_id)
account = BankAccount.from_events(events)
return account.balance if account else 0
# Usage
store = EventStore()
service = BankAccountService(store)
# All operations stored as events
service.open_account("ACC-001", 1000)
service.deposit("ACC-001", 500)
service.withdraw("ACC-001", 200)
# Balance reconstructed from events
balance = service.get_balance("ACC-001") # 1300
# Full audit trail available
events = store.get_events("ACC-001")
for event in events:
print(f"{event.type}: {event.data}")Event Sourcing Benefits: Complete audit trail, time travel (replay events), event replay for debugging, multiple read models from same events.
When to Use Event-Driven Architecture
- Loose coupling needed: Services should not directly depend on each other
- Real-time processing: React to changes as they happen
- Asynchronous workflows: Long-running processes, background tasks
- Multiple consumers: Same event triggers different actions in different services
- Scalability requirements: Need to handle high throughput with decoupled components
- Audit and compliance: Need complete event history
Pros & Cons
✅ Pros
- Loose coupling: Services don't need to know about each other
- Scalability: Easy to scale event consumers independently
- Flexibility: Add new event consumers without changing producers
- Resilience: Failures in one consumer don't affect others
- Real-time processing: React to events as they happen
- Audit trail: All events are recorded for compliance
❌ Cons
- Complexity: Harder to understand flow than synchronous calls
- Debugging difficulty: Tracing events across services is challenging
- Eventual consistency: No immediate guarantees of data consistency
- Message ordering: Ensuring correct event order can be difficult
- Infrastructure overhead: Need message brokers, monitoring tools
- Testing challenges: Harder to write integration tests
💡 Pro Tip: Start with synchronous communication and introduce events only when you need loose coupling, asynchronous processing, or multiple consumers. Event-driven architecture adds complexity, so make sure the benefits justify the cost.
CQRS (Command Query Responsibility Segregation)
Intent: Separate read and write operations into different models. Commands (writes) change state, queries (reads) return data. This allows optimizing each side independently and often pairs with Event Sourcing.
Traditional vs CQRS
# Traditional: Same model for reads and writes
class TraditionalOrderService:
def create_order(self, user_id, items):
# Write to database
order = Order.objects.create(user_id=user_id, items=items)
return order
def get_order(self, order_id):
# Read from same database/model
return Order.objects.get(id=order_id)
def get_user_orders(self, user_id):
# Complex query on write-optimized schema
return Order.objects.filter(user_id=user_id).join(OrderItems)
# CQRS: Separate command and query sides
# COMMAND SIDE - Optimized for writes
class CreateOrderCommand:
def __init__(self, user_id, items):
self.user_id = user_id
self.items = items
class CancelOrderCommand:
def __init__(self, order_id):
self.order_id = order_id
class CommandHandler:
def __init__(self, event_store, event_bus):
self.event_store = event_store
self.event_bus = event_bus
def handle_create_order(self, cmd: CreateOrderCommand):
# Validate command
if not cmd.items:
raise ValueError("Order must have items")
# Create event
event = Event("order.created", {
"order_id": str(uuid.uuid4()),
"user_id": cmd.user_id,
"items": cmd.items,
"total": sum(item.price * item.qty for item in cmd.items)
})
# Store event
self.event_store.append(event)
# Publish to update read models
self.event_bus.publish(event)
return event.data["order_id"]
def handle_cancel_order(self, cmd: CancelOrderCommand):
event = Event("order.cancelled", {
"order_id": cmd.order_id
})
self.event_store.append(event)
self.event_bus.publish(event)
# QUERY SIDE - Optimized for reads
class OrderReadModel:
"""Denormalized, read-optimized data structure"""
def __init__(self):
self.orders = {} # In reality: Redis, Elasticsearch, etc.
def get_order(self, order_id):
"""Fast read from denormalized store"""
return self.orders.get(order_id)
def get_user_orders(self, user_id):
"""Pre-computed query result"""
return [
order for order in self.orders.values()
if order["user_id"] == user_id
]
def get_order_summary(self, order_id):
"""Complex aggregation pre-computed"""
order = self.orders.get(order_id)
if not order:
return None
return {
"order_id": order_id,
"total": order["total"],
"item_count": len(order["items"]),
"status": order["status"]
}
# Projection: Updates read model from events
class OrderProjection:
def __init__(self, read_model: OrderReadModel, event_bus):
self.read_model = read_model
event_bus.subscribe("order.created", self.on_order_created)
event_bus.subscribe("order.cancelled", self.on_order_cancelled)
def on_order_created(self, event: Event):
"""Update read model when order created"""
self.read_model.orders[event.data["order_id"]] = {
"order_id": event.data["order_id"],
"user_id": event.data["user_id"],
"items": event.data["items"],
"total": event.data["total"],
"status": "pending"
}
def on_order_cancelled(self, event: Event):
"""Update read model when order cancelled"""
order_id = event.data["order_id"]
if order_id in self.read_model.orders:
self.read_model.orders[order_id]["status"] = "cancelled"
# Usage
event_store = EventStore()
event_bus = EventBus()
command_handler = CommandHandler(event_store, event_bus)
read_model = OrderReadModel()
projection = OrderProjection(read_model, event_bus)
# Write path
create_cmd = CreateOrderCommand(
user_id="user_123",
items=[{"product_id": "prod_1", "price": 29.99, "qty": 2}]
)
order_id = command_handler.handle_create_order(create_cmd)
# Read path - different model, optimized for queries
order = read_model.get_order(order_id)
user_orders = read_model.get_user_orders("user_123")
summary = read_model.get_order_summary(order_id)Advanced CQRS: Multiple Read Models
# Different read models for different use cases
class OrderListReadModel:
"""Optimized for listing orders"""
def __init__(self):
self.orders_by_user = defaultdict(list)
def get_user_orders(self, user_id):
return self.orders_by_user[user_id]
class OrderAnalyticsReadModel:
"""Optimized for analytics/reporting"""
def __init__(self):
self.daily_revenue = defaultdict(float)
self.product_sales = defaultdict(int)
def get_revenue_by_date(self, date):
return self.daily_revenue[date]
def get_top_products(self, limit=10):
return sorted(
self.product_sales.items(),
key=lambda x: x[1],
reverse=True
)[:limit]
class OrderSearchReadModel:
"""Optimized for full-text search"""
def __init__(self):
self.search_index = {} # Elasticsearch in reality
def search_orders(self, query):
# Full-text search capabilities
return [
order for order in self.search_index.values()
if query.lower() in str(order).lower()
]
# Each projection updates its specialized read model
class AnalyticsProjection:
def __init__(self, analytics_model, event_bus):
self.analytics = analytics_model
event_bus.subscribe("order.created", self.on_order_created)
def on_order_created(self, event: Event):
date = datetime.now().date()
self.analytics.daily_revenue[date] += event.data["total"]
for item in event.data["items"]:
self.analytics.product_sales[item["product_id"]] += item["qty"]
# Benefit: Scale read and write sides independently
# - Write side: Strong consistency, ACID transactions
# - Read side: Eventually consistent, highly scalableWhen to Use CQRS
- Different read/write performance needs: Writes need ACID, reads need speed
- Multiple read models: Need different views of the same data (list, search, analytics)
- High read-to-write ratio: Many more reads than writes
- Complex domain logic: Business rules on write side, simple projections on read side
- Scalability requirements: Need to scale reads and writes independently
- Event Sourcing: Often paired with Event Sourcing for complete audit trail
Pros & Cons
✅ Pros
- Independent scaling: Scale read and write sides separately
- Optimized queries: Denormalized read models for fast queries
- Multiple views: Different read models for different use cases
- Performance: Read queries don't impact write operations
- Simplified queries: No complex joins needed in read models
- Flexibility: Add new read models without changing write side
❌ Cons
- Complexity: Two separate models to maintain instead of one
- Eventual consistency: Read models may be slightly out of date
- Data duplication: Same data stored in multiple places
- Synchronization overhead: Need projections to keep read models updated
- Learning curve: More difficult concept than traditional CRUD
- Infrastructure: Requires message bus and multiple data stores
⚠️ Warning: CQRS adds significant complexity. Only use it when you have a clear need for independent scaling, optimized queries, or multiple read models. For most applications, a traditional unified model is simpler and sufficient.
Event Sourcing
Intent: Store the state of your application as a sequence of events rather than the current state. Instead of UPDATE statements, you append events. The current state is derived by replaying events. This provides a complete audit log and enables time travel.
💡 Key Insight: Event Sourcing is often combined with CQRS. Events become the write model, and you build optimized read models by projecting events.
Core Concepts
# Traditional approach - store current state
class BankAccount:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0 # Current state only
def deposit(self, amount):
self.balance += amount # Lost history - we don't know how we got here
db.update(account_id, balance=self.balance)
# Event Sourcing - store events
class AccountCreated:
def __init__(self, account_id, owner):
self.account_id = account_id
self.owner = owner
self.timestamp = datetime.now()
class MoneyDeposited:
def __init__(self, account_id, amount):
self.account_id = account_id
self.amount = amount
self.timestamp = datetime.now()
class MoneyWithdrawn:
def __init__(self, account_id, amount):
self.account_id = account_id
self.amount = amount
self.timestamp = datetime.now()
# Event Store - append-only log
class EventStore:
def __init__(self):
self.events = [] # In reality: database or Kafka
def append(self, event):
"""Append event - never update or delete"""
self.events.append(event)
print(f"Event stored: {event.__class__.__name__}")
def get_events(self, aggregate_id):
"""Get all events for an aggregate"""
return [e for e in self.events if e.account_id == aggregate_id]
# Rebuild state by replaying events
class BankAccountEventSourced:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0
self.owner = None
self.version = 0
def apply_event(self, event):
"""Apply event to modify state"""
if isinstance(event, AccountCreated):
self.owner = event.owner
elif isinstance(event, MoneyDeposited):
self.balance += event.amount
elif isinstance(event, MoneyWithdrawn):
self.balance -= event.amount
self.version += 1
def load_from_history(self, events):
"""Rebuild current state from events"""
for event in events:
self.apply_event(event)
def deposit(self, amount):
"""Create event instead of modifying state directly"""
event = MoneyDeposited(self.account_id, amount)
event_store.append(event)
self.apply_event(event)
def withdraw(self, amount):
if self.balance >= amount:
event = MoneyWithdrawn(self.account_id, amount)
event_store.append(event)
self.apply_event(event)
else:
raise ValueError("Insufficient funds")
# Usage
event_store = EventStore()
# Create account
created_event = AccountCreated("ACC-001", "Alice")
event_store.append(created_event)
account = BankAccountEventSourced("ACC-001")
account.load_from_history(event_store.get_events("ACC-001"))
# Perform operations - each creates an event
account.deposit(100)
account.deposit(50)
account.withdraw(30)
print(f"Current balance: {account.balance}") # 120
# Time travel - rebuild state at any point in time
past_account = BankAccountEventSourced("ACC-001")
past_events = event_store.get_events("ACC-001")[:2] # First 2 events only
past_account.load_from_history(past_events)
print(f"Balance after first deposit: {past_account.balance}") # 100Projection - Building Read Models
# Read model - optimized for queries
class AccountSummary:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0
self.total_deposits = 0
self.total_withdrawals = 0
self.transaction_count = 0
# Projection - converts events to read model
class AccountSummaryProjection:
def __init__(self):
self.summaries = {}
def handle_event(self, event):
"""Update read model when event occurs"""
account_id = event.account_id
if account_id not in self.summaries:
self.summaries[account_id] = AccountSummary(account_id)
summary = self.summaries[account_id]
if isinstance(event, MoneyDeposited):
summary.balance += event.amount
summary.total_deposits += event.amount
summary.transaction_count += 1
elif isinstance(event, MoneyWithdrawn):
summary.balance -= event.amount
summary.total_withdrawals += event.amount
summary.transaction_count += 1
def get_summary(self, account_id):
return self.summaries.get(account_id)
# Projection subscribes to events
projection = AccountSummaryProjection()
# Process events to build read model
for event in event_store.get_events("ACC-001"):
projection.handle_event(event)
# Query read model - fast lookups
summary = projection.get_summary("ACC-001")
print(f"Total deposits: {summary.total_deposits}")
print(f"Transaction count: {summary.transaction_count}")Snapshots - Performance Optimization
# Problem: Replaying 1 million events is slow!
# Solution: Snapshots - periodic state checkpoints
class Snapshot:
def __init__(self, aggregate_id, state, version):
self.aggregate_id = aggregate_id
self.state = state
self.version = version
self.timestamp = datetime.now()
class SnapshotStore:
def __init__(self):
self.snapshots = {}
def save_snapshot(self, snapshot):
self.snapshots[snapshot.aggregate_id] = snapshot
def get_snapshot(self, aggregate_id):
return self.snapshots.get(aggregate_id)
class BankAccountWithSnapshots(BankAccountEventSourced):
def save_snapshot(self, snapshot_store):
"""Create snapshot of current state"""
snapshot = Snapshot(
self.account_id,
{"balance": self.balance, "owner": self.owner},
self.version
)
snapshot_store.save_snapshot(snapshot)
def load_from_snapshot(self, snapshot, event_store):
"""Fast load: snapshot + events since snapshot"""
# Restore from snapshot
self.balance = snapshot.state["balance"]
self.owner = snapshot.state["owner"]
self.version = snapshot.version
# Replay only events since snapshot
recent_events = [
e for e in event_store.get_events(self.account_id)
if e.timestamp > snapshot.timestamp
]
self.load_from_history(recent_events)
# Every 100 events, create a snapshot
snapshot_store = SnapshotStore()
if account.version % 100 == 0:
account.save_snapshot(snapshot_store)
# Fast load using snapshot
new_account = BankAccountWithSnapshots("ACC-001")
snapshot = snapshot_store.get_snapshot("ACC-001")
if snapshot:
new_account.load_from_snapshot(snapshot, event_store)
else:
new_account.load_from_history(event_store.get_events("ACC-001"))When to Use Event Sourcing
- Audit requirements: Financial systems, healthcare, legal - need complete history
- Time travel: Need to see state at any point in time
- Debugging: Replay production events to reproduce bugs
- Analytics: Run new queries against historical data
- Event-driven systems: Already using events for communication
✅ Pros
- Complete audit trail
- Time travel capabilities
- Never lose data - append only
- Easy to add new projections later
- Natural fit for event-driven systems
- Enables CQRS effectively
❌ Cons
- Learning curve - different mindset
- Event schema evolution is hard
- Eventually consistent
- Queries require projections
- Can't delete data easily (GDPR concern)
- Storage grows continuously
⚠️ Warning: Event Sourcing adds significant complexity. Only use it when you have a clear need for the benefits it provides. Most applications don't need event sourcing.
Hexagonal Architecture (Ports & Adapters)
Intent: Isolate core business logic from external concerns (databases, UI, APIs). The core defines "ports" (interfaces), and external systems provide "adapters" (implementations). This makes the system testable and technology-agnostic.
Architecture Layers
# CORE DOMAIN - No external dependencies
class Order:
"""Pure domain model - no framework dependencies"""
def __init__(self, order_id, user_id, items):
self.id = order_id
self.user_id = user_id
self.items = items
self.status = "pending"
self.total = sum(item.price * item.quantity for item in items)
def cancel(self):
"""Business logic in domain model"""
if self.status == "shipped":
raise ValueError("Cannot cancel shipped order")
self.status = "cancelled"
def mark_as_paid(self):
if self.status != "pending":
raise ValueError("Order already processed")
self.status = "paid"
# PORTS - Interfaces that core depends on (dependency inversion)
class OrderRepository(ABC):
"""Port: Interface for persistence"""
@abstractmethod
def save(self, order: Order):
pass
@abstractmethod
def find_by_id(self, order_id: str) -> Order:
pass
@abstractmethod
def find_by_user(self, user_id: str) -> list[Order]:
pass
class PaymentGateway(ABC):
"""Port: Interface for payment processing"""
@abstractmethod
def charge(self, amount: float, user_id: str) -> bool:
pass
class NotificationService(ABC):
"""Port: Interface for notifications"""
@abstractmethod
def send_order_confirmation(self, order: Order):
pass
# APPLICATION SERVICES - Use cases/business logic
class OrderService:
"""Core business logic - depends only on ports, not implementations"""
def __init__(
self,
order_repo: OrderRepository,
payment_gateway: PaymentGateway,
notification: NotificationService
):
# Dependency injection of interfaces
self.orders = order_repo
self.payment = payment_gateway
self.notifications = notification
def create_order(self, user_id: str, items: list) -> Order:
"""Use case: Create and process order"""
# Create domain model
order = Order(
order_id=str(uuid.uuid4()),
user_id=user_id,
items=items
)
# Use ports (interfaces) - don't know implementation
if not self.payment.charge(order.total, user_id):
raise ValueError("Payment failed")
order.mark_as_paid()
self.orders.save(order)
self.notifications.send_order_confirmation(order)
return order
def cancel_order(self, order_id: str):
"""Use case: Cancel order"""
order = self.orders.find_by_id(order_id)
if not order:
raise ValueError("Order not found")
order.cancel() # Business logic in domain
self.orders.save(order)
# ADAPTERS - Concrete implementations of ports
# Database Adapter
class PostgreSQLOrderRepository(OrderRepository):
"""Adapter: PostgreSQL implementation"""
def __init__(self, connection):
self.db = connection
def save(self, order: Order):
self.db.execute(
"INSERT INTO orders VALUES (%s, %s, %s, %s)",
(order.id, order.user_id, order.status, order.total)
)
def find_by_id(self, order_id: str) -> Order:
row = self.db.query("SELECT * FROM orders WHERE id = %s", order_id)
return self._to_domain(row)
def find_by_user(self, user_id: str) -> list[Order]:
rows = self.db.query("SELECT * FROM orders WHERE user_id = %s", user_id)
return [self._to_domain(row) for row in rows]
def _to_domain(self, row) -> Order:
# Map database row to domain model
return Order(row["id"], row["user_id"], row["items"])
class MongoOrderRepository(OrderRepository):
"""Adapter: MongoDB implementation - easily swappable"""
def __init__(self, mongo_client):
self.collection = mongo_client.orders
def save(self, order: Order):
self.collection.insert_one({
"_id": order.id,
"user_id": order.user_id,
"status": order.status,
"total": order.total
})
def find_by_id(self, order_id: str) -> Order:
doc = self.collection.find_one({"_id": order_id})
return self._to_domain(doc)
def find_by_user(self, user_id: str) -> list[Order]:
docs = self.collection.find({"user_id": user_id})
return [self._to_domain(doc) for doc in docs]
def _to_domain(self, doc) -> Order:
return Order(doc["_id"], doc["user_id"], doc.get("items", []))
# Payment Adapter
class StripePaymentGateway(PaymentGateway):
"""Adapter: Stripe implementation"""
def charge(self, amount: float, user_id: str) -> bool:
try:
stripe.charge(amount=int(amount * 100), customer=user_id)
return True
except stripe.error.CardError:
return False
class MockPaymentGateway(PaymentGateway):
"""Adapter: Mock for testing"""
def charge(self, amount: float, user_id: str) -> bool:
return True # Always succeeds in tests
# Notification Adapter
class EmailNotificationService(NotificationService):
"""Adapter: Email implementation"""
def send_order_confirmation(self, order: Order):
send_email(
to=get_user_email(order.user_id),
subject="Order Confirmation",
body=f"Your order {order.id} has been confirmed!"
)
# COMPOSITION - Wire it all together at startup
# Production configuration
order_service = OrderService(
order_repo=PostgreSQLOrderRepository(db_connection),
payment_gateway=StripePaymentGateway(),
notification=EmailNotificationService()
)
# Test configuration - easily swap adapters
test_order_service = OrderService(
order_repo=InMemoryOrderRepository(),
payment_gateway=MockPaymentGateway(),
notification=MockNotificationService()
)
# Core business logic never changes, only adapters swapWhen to Use Hexagonal Architecture
- Complex business logic: Core domain rules are complex and need protection from infrastructure changes
- Multiple interfaces: Need to support web, CLI, mobile, API simultaneously
- Technology flexibility: Want to swap databases, frameworks, or third-party services easily
- High testability: Need to test business logic in isolation without external dependencies
- Long-lived systems: Application will outlive its framework and infrastructure
- Domain-Driven Design: Works well with DDD to protect domain models
Pros & Cons
✅ Pros
- Testability: Business logic testable without infrastructure
- Framework independence: Core logic not coupled to frameworks
- Technology agnostic: Easy to swap databases, UI, APIs
- Clear boundaries: Explicit separation between business and technical concerns
- Flexibility: Multiple adapters for same port (e.g., SQL/NoSQL)
- Maintainability: Changes to infrastructure don't affect business logic
❌ Cons
- Increased complexity: More abstractions, interfaces, and indirection
- Boilerplate code: Many interfaces and adapter implementations
- Learning curve: Harder to understand than straightforward layered architecture
- Overhead for simple apps: Not justified for CRUD-heavy applications
- Performance impact: Extra layers of abstraction may reduce performance
- Over-engineering risk: Easy to create unnecessary abstractions
💡 Pro Tip: Hexagonal architecture shines when business logic is complex and needs protection from infrastructure changes. For simple CRUD apps, the extra abstraction may not be worth the cost. Start simpler and evolve to hexagonal when the benefits justify the complexity.
Layered Architecture
Intent: Organize code into horizontal layers where each layer depends only on layers below it. Common layers: Presentation → Business Logic → Data Access. Simple and well-understood, though can lead to tight coupling.
# Classic 3-tier layered architecture
# LAYER 1: PRESENTATION (Controllers/Views)
class OrderController:
def __init__(self, order_service):
self.order_service = order_service # Depends on layer below
def create_order(self, request):
"""Handle HTTP request"""
user_id = request.user_id
items = request.json["items"]
# Delegate to business layer
order = self.order_service.create_order(user_id, items)
# Return HTTP response
return {"order_id": order.id, "status": order.status}
# LAYER 2: BUSINESS LOGIC (Services)
class OrderService:
def __init__(self, order_repository, payment_service):
self.orders = order_repository # Depends on layer below
self.payments = payment_service
def create_order(self, user_id, items):
"""Business logic - no HTTP/database details"""
# Validate business rules
if not items:
raise ValueError("Order must have items")
# Create order
order = Order(user_id, items)
# Process payment
if not self.payments.process(order.total):
raise ValueError("Payment failed")
# Save to database via repository
self.orders.save(order)
return order
# LAYER 3: DATA ACCESS (Repositories)
class OrderRepository:
def __init__(self, database):
self.db = database # Depends on infrastructure
def save(self, order):
"""Pure data access - no business logic"""
self.db.execute(
"INSERT INTO orders (id, user_id, total) VALUES (%s, %s, %s)",
(order.id, order.user_id, order.total)
)
def find_by_id(self, order_id):
row = self.db.query("SELECT * FROM orders WHERE id = %s", order_id)
return self._to_order(row)
# Dependencies flow downward only
# Presentation → Business → Data
# Each layer only knows about the layer directly below itWhen to Use Layered Architecture
- Traditional web applications: Standard CRUD apps with database and UI
- Team familiarity: Most developers understand layers, easy onboarding
- Clear separation needed: Want to separate presentation, business, and data concerns
- Standard technology stack: Using conventional frameworks (Spring, .NET, Rails)
- Simple to moderate complexity: Not overly complex business logic
- Quick development: Need to build something straightforward quickly
Pros & Cons
✅ Pros
- Simplicity: Easy to understand and implement
- Well-known pattern: Most developers are familiar with it
- Clear organization: Code organized by technical function
- Tool support: Most frameworks encourage layered architecture
- Separation of concerns: Different responsibilities in different layers
- Easy to start: Low initial complexity, quick to build
❌ Cons
- Tight coupling: Upper layers tightly coupled to lower layers
- Database-centric: Often becomes database-driven design
- Logic leakage: Business logic tends to leak into other layers
- Testing difficulty: Hard to test without database/UI
- Pass-through layers: Layers with no real logic, just delegation
- Inflexibility: Difficult to swap implementations (e.g., database)
⚠️ Warning: Layered architecture can lead to tightly coupled systems where business logic leaks into presentation and data layers. For complex business logic, consider hexagonal or clean architecture for better separation and testability.
Saga Pattern
Intent: Manage distributed transactions across multiple services. Instead of a single ACID transaction, a saga is a sequence of local transactions where each service publishes events to trigger the next step. Compensating transactions handle failures.
Choreography-Based Saga
In choreography, services listen to events and decide what to do next. There's no central coordinator. Each service publishes events that trigger the next step in other services. This keeps services loosely coupled but can make the workflow harder to understand.
# Choreography-based Saga: Services react to events
class OrderService:
def create_order(self, user_id, items):
order = Order(user_id, items, status="pending")
self.db.save(order)
# Publish event to trigger next step
event_bus.publish("order.created", {
"order_id": order.id,
"user_id": user_id,
"items": items,
"total": order.total
})
return order
@event_bus.subscribe("payment.failed")
def on_payment_failed(self, event):
"""Compensating transaction"""
order_id = event["order_id"]
order = self.db.get(order_id)
order.status = "cancelled"
self.db.save(order)
event_bus.publish("order.cancelled", {"order_id": order_id})
class PaymentService:
@event_bus.subscribe("order.created")
def process_payment(self, event):
order_id = event["order_id"]
user_id = event["user_id"]
amount = event["total"]
try:
payment = self.charge(user_id, amount)
event_bus.publish("payment.completed", {
"order_id": order_id,
"payment_id": payment.id
})
except PaymentError:
event_bus.publish("payment.failed", {
"order_id": order_id
})
@event_bus.subscribe("order.cancelled")
def refund_payment(self, event):
"""Compensating transaction"""
order_id = event["order_id"]
payment = self.db.get_by_order(order_id)
if payment:
self.refund(payment.id)
class InventoryService:
@event_bus.subscribe("payment.completed")
def reserve_inventory(self, event):
order_id = event["order_id"]
items = self.get_order_items(order_id)
try:
for item in items:
self.reserve_stock(item.product_id, item.quantity)
event_bus.publish("inventory.reserved", {
"order_id": order_id
})
except InsufficientStockError:
event_bus.publish("inventory.reservation_failed", {
"order_id": order_id
})
@event_bus.subscribe("order.cancelled")
def release_inventory(self, event):
"""Compensating transaction"""
order_id = event["order_id"]
items = self.get_order_items(order_id)
for item in items:
self.release_stock(item.product_id, item.quantity)
class ShippingService:
@event_bus.subscribe("inventory.reserved")
def create_shipment(self, event):
order_id = event["order_id"]
shipment = self.create_shipping_label(order_id)
event_bus.publish("shipment.created", {
"order_id": order_id,
"tracking_number": shipment.tracking
})💡 Choreography Explained: Notice how each service only listens to events and publishes new events. No service knows about the overall workflow. Order → Payment → Inventory → Shipping happens through event reactions, not a central plan. This is highly decoupled but harder to visualize the complete flow.
Orchestration-Based Saga
In orchestration, a central coordinator (the orchestrator) explicitly calls each service in sequence. The orchestrator knows the complete workflow and handles compensation if any step fails. This makes the flow explicit and easier to understand, but creates a single point of coordination.
# Orchestration-based Saga: Central coordinator
class OrderSagaOrchestrator:
"""Centralized saga coordination"""
def __init__(self, payment_svc, inventory_svc, shipping_svc):
self.payment = payment_svc
self.inventory = inventory_svc
self.shipping = shipping_svc
async def execute_order_saga(self, order):
"""Execute saga steps in sequence"""
try:
# Step 1: Payment
payment = await self.payment.charge(order.user_id, order.total)
# Step 2: Reserve inventory
await self.inventory.reserve(order.items)
# Step 3: Create shipment
await self.shipping.create(order.id)
return {"status": "success"}
except PaymentError:
# No compensation needed - first step failed
return {"status": "payment_failed"}
except InsufficientStockError:
# Compensate: Refund payment
await self.payment.refund(payment.id)
return {"status": "out_of_stock"}
except ShippingError:
# Compensate: Release inventory and refund payment
await self.inventory.release(order.items)
await self.payment.refund(payment.id)
return {"status": "shipping_failed"}✅ Orchestration Benefits: The complete workflow is visible in one place (the orchestrator). You can see exactly what steps happen and in what order. Compensation logic is explicit with try/catch blocks. This makes debugging and monitoring much easier than choreography.
Advanced: Saga State Machine
For production systems, you need to track saga state persistently. If the orchestrator crashes mid-saga, it should resume from where it left off. This requires a state machine that records each step and its compensation actions.
# Saga execution state machine
class SagaState:
STARTED = "started"
PAYMENT_COMPLETED = "payment_completed"
INVENTORY_RESERVED = "inventory_reserved"
SHIPMENT_CREATED = "shipment_created"
COMPLETED = "completed"
FAILED = "failed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
class OrderSaga:
def __init__(self, order_id):
self.order_id = order_id
self.state = SagaState.STARTED
self.compensation_steps = []
def execute(self):
try:
self._step_payment()
self._step_inventory()
self._step_shipping()
self.state = SagaState.COMPLETED
except Exception as e:
self._compensate()
raise
def _step_payment(self):
# Execute payment
self.compensation_steps.append(self._compensate_payment)
self.state = SagaState.PAYMENT_COMPLETED
def _compensate(self):
"""Execute compensation steps in reverse order"""
self.state = SagaState.COMPENSATING
for compensate_fn in reversed(self.compensation_steps):
try:
compensate_fn()
except Exception as e:
logging.error(f"Compensation failed: {e}")
self.state = SagaState.COMPENSATED| Feature | Choreography | Orchestration |
|---|---|---|
| Complexity | Low (simple flows) | High (complex state) |
| Coupling | Loose | Tight (to Orchestrator) |
| Visibility | Difficult to track | Centralized & Clear |
Architect's Tip: Use Choreography for simple workflows with 2-3 steps to keep services decoupled. Switch to Orchestration for complex business processes (like flight + hotel + car bookings) to prevent "spaghetti events."
Saga Trade-offs:
- Choreography: Decentralized, services autonomous, but harder to track overall flow.
- Orchestration: Centralized control, easier to understand, but orchestrator is single point of failure.
Cloud-Native Implementation
Modern cloud providers offer Managed Orchestrators. These tools handle the "hard parts" of Sagas: state persistence, exponential backoff retries, and ensuring compensating transactions run even if a server crashes.
AWS Step Functions
A JSON-based state machine (ASL) that orchestrates Lambda functions with built-in retry/catch logic.
- Visual Workflow UI
- Pay-per-transition pricing
- Best for high-volume serverless
Azure Durable Functions
Write your Saga Orchestrator directly in Python or C# using async/await patterns.
- Code-first approach
- Automatic checkpointing
- Best for complex conditional logic
Declarative JSON (Amazon States Language)
{
"StartAt": "ProcessPayment",
"States": {
"ProcessPayment": {
"Type": "Task",
"Resource": "arn:aws:lambda:payment",
"Next": "ReserveInventory",
"Catch": [{
"ErrorEquals": ["PaymentError"],
"Next": "CancelOrder"
}]
},
"ReserveInventory": {
"Type": "Task",
"Resource": "arn:aws:lambda:inventory",
"Next": "Success",
"Catch": [{
"ErrorEquals": ["OutOfStock"],
"Next": "RefundPayment"
}]
},
"RefundPayment": {
"Type": "Task",
"Resource": "arn:aws:lambda:refund",
"Next": "CancelOrder"
},
"CancelOrder": { "Type": "Fail" },
"Success": { "Type": "Succeed" }
}
}Imperative Code-First (Python)
import azure.durable_functions as df
def orchestrator(context: df.DurableOrchestrationContext):
data = context.get_input()
try:
# Step 1: Charge
yield context.call_activity("Charge", data)
try:
# Step 2: Inventory
yield context.call_activity("Reserve", data)
except Exception:
# Compensate Step 1
yield context.call_activity("Refund", data)
return "Rolled Back"
return "Complete"
except Exception:
return "Failed at Start"Why use Managed Orchestrators?
Persistence
The workflow state is saved to disk. If the orchestrator restarts, it picks up exactly where it left off.
Auto-Retry
Configurable retry policies (e.g., "retry 3 times with 10s backoff") without writing a single line of logic.
Observability
Visual tools show you exactly which "box" in the saga failed, making debugging distributed systems much easier.
When to Use Saga Pattern
- Distributed transactions: Business process spans multiple microservices
- No 2PC available: Can't use traditional two-phase commit
- Long-running workflows: Process involves steps that take time (e.g., external APIs)
- Compensatable steps: Each step has a compensating action to undo it
- Eventual consistency acceptable: Don't need immediate consistency across all services
- Complex business workflows: Order processing, booking systems, payment flows
Pros & Cons
✅ Pros
- Distributed transactions: Coordinate across multiple services
- No distributed locks: Avoids two-phase commit complexity
- Resilience: Handles failures with compensating transactions
- Service autonomy: Each service manages its own data
- Scalability: No central lock coordinator bottleneck
- Long-running processes: Supports workflows that take hours/days
❌ Cons
- Complexity: Much more complex than ACID transactions
- Compensating logic: Must implement undo operations for each step
- Eventual consistency: Intermediate states visible to users
- Testing difficulty: Hard to test all failure scenarios
- Debugging challenges: Tracing saga execution across services
- Orchestrator maintenance: Central point of coordination to manage
⚠️ Important: Sagas are significantly more complex than monolithic ACID transactions. Only use them when you genuinely need to coordinate transactions across multiple services. If you can keep related data in one service, prefer that approach with traditional database transactions.
Database per Service
Intent: Each microservice owns its database and no other service can access it directly. Services communicate only through APIs. This enables independent development, deployment, and technology choices, but introduces data consistency challenges.
💡 Core Principle: Services are loosely coupled at the data level. You can't just JOIN across services - you need sagas, event-driven synchronization, or API calls.
Structure
# Each service has its own database
user_service/
├── user_api.py
├── user_db.py # User database - PostgreSQL
└── models/
└── user.py
order_service/
├── order_api.py
├── order_db.py # Order database - PostgreSQL
└── models/
└── order.py
# NOTE: No user table! Only user_id reference
inventory_service/
├── inventory_api.py
├── inventory_db.py # Inventory database - MongoDB (different tech!)
└── models/
└── product.py
# Each database is isolated
# ✅ user_service CAN access user_db
# ❌ order_service CANNOT access user_db
# ✅ order_service must call user_service API to get user dataChallenge: Querying Across Services
# Monolithic approach - easy JOIN
def get_order_details_monolith(order_id):
result = db.query("""
SELECT o.*, u.name, u.email, p.name as product_name
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN products p ON o.product_id = p.id
WHERE o.id = ?
""", order_id)
return result
# Microservices approach - multiple API calls
class OrderService:
def get_order_details(self, order_id):
# Get order from our database
order = self.db.orders.get(order_id)
# Call user service API for user data
user_response = requests.get(f"{USER_SERVICE_URL}/users/{order.user_id}")
user = user_response.json()
# Call inventory service API for product data
product_response = requests.get(f"{INVENTORY_SERVICE_URL}/products/{order.product_id}")
product = product_response.json()
# Assemble the complete response
return {
"order_id": order.id,
"total": order.total,
"status": order.status,
"user": {
"name": user["name"],
"email": user["email"]
},
"product": {
"name": product["name"],
"price": product["price"]
}
}
# Problem: Multiple network calls, slower than JOIN
# Solution options:
# 1. Accept the performance hit (it's usually fine)
# 2. Use CQRS with denormalized read models
# 3. Create a dedicated API Gateway/BFF to aggregate
# 4. Event-driven data replication (eventual consistency)Pattern: Data Replication for Reads
# Order service maintains a cached copy of user data
class OrderService:
def __init__(self, event_bus):
self.db = OrderDatabase()
self.user_cache = {} # Local cache of user data
# Subscribe to user events for cache invalidation
event_bus.subscribe("user.created", self.on_user_created)
event_bus.subscribe("user.updated", self.on_user_updated)
def on_user_created(self, event):
"""Update local cache when user is created"""
self.user_cache[event["user_id"]] = {
"name": event["name"],
"email": event["email"]
}
def on_user_updated(self, event):
"""Update local cache when user is updated"""
self.user_cache[event["user_id"]] = {
"name": event["name"],
"email": event["email"]
}
def get_order_details(self, order_id):
"""Fast lookup - no external API calls needed"""
order = self.db.orders.get(order_id)
user = self.user_cache.get(order.user_id)
return {
"order_id": order.id,
"total": order.total,
"user": user # From cache, not API call
}
# Trade-off: Eventual consistency
# The cache might be slightly stale, but queries are fastTechnology Diversity
Each service can choose the best database for its needs:
# User Service - Relational data
user_service:
database: PostgreSQL
reason: ACID transactions, relationships
# Product Service - Document storage
product_service:
database: MongoDB
reason: Flexible schema, nested documents
# Analytics Service - Column store
analytics_service:
database: ClickHouse
reason: Fast aggregations on large datasets
# Session Service - Key-value
session_service:
database: Redis
reason: High-speed caching, TTL support
# Search Service - Full-text search
search_service:
database: Elasticsearch
reason: Powerful search capabilitiesWhen to Use Database per Service
- Microservices architecture: Essential for true service independence
- Team autonomy: Teams can evolve databases independently
- Different data needs: Services benefit from different database technologies
- Independent scaling: Scale service data storage separately
✅ Pros
- True service independence
- Technology diversity - right tool for job
- Independent scaling
- Isolated failures
- Team autonomy
- Easier to evolve schema
❌ Cons
- No cross-service transactions
- No cross-service JOINs
- Data consistency challenges
- Requires sagas for distributed transactions
- Operational complexity - many databases
- Potential data duplication
⚠️ Warning: Don't start with Database per Service. Begin with a shared database or modular monolith. Only split databases when services are stable and boundaries are clear. It's much easier to split databases later than to merge them!
API Gateway Pattern
Intent: Provide a single entry point for clients to access multiple microservices. The gateway handles routing, authentication, rate limiting, and can aggregate results from multiple services.
class APIGateway:
def __init__(self):
self.user_service = HTTPClient("http://user-service")
self.order_service = HTTPClient("http://order-service")
self.product_service = HTTPClient("http://product-service")
self.auth_service = HTTPClient("http://auth-service")
@app.route("/api/user-dashboard")
async def get_user_dashboard(self, request):
"""Aggregate data from multiple services"""
user_id = request.user_id
# Authenticate
if not await self._authenticate(request.token):
return {"error": "Unauthorized"}, 401
# Rate limiting
if not await self._check_rate_limit(user_id):
return {"error": "Rate limit exceeded"}, 429
# Parallel requests to multiple services
user_data, orders, recommendations = await asyncio.gather(
self.user_service.get(f"/users/{user_id}"),
self.order_service.get(f"/orders?user_id={user_id}"),
self.product_service.get(f"/recommendations/{user_id}")
)
# Aggregate response
return {
"user": user_data,
"recent_orders": orders[:5],
"recommended_products": recommendations
}
async def _authenticate(self, token):
"""Centralized authentication"""
response = await self.auth_service.post("/validate", {"token": token})
return response.get("valid", False)
async def _check_rate_limit(self, user_id):
"""Rate limiting"""
key = f"rate_limit:{user_id}"
count = await redis.incr(key)
if count == 1:
await redis.expire(key, 60) # 1 minute window
return count <= 100 # 100 requests per minute
# Backend for Frontend (BFF) Pattern
class MobileAPIGateway:
"""Optimized for mobile clients"""
@app.route("/mobile/dashboard")
async def get_mobile_dashboard(self, request):
# Minimal data for mobile bandwidth
return {
"user": await self._get_minimal_user(request.user_id),
"orders": await self._get_order_summaries(request.user_id)
}
class WebAPIGateway:
"""Optimized for web clients"""
@app.route("/web/dashboard")
async def get_web_dashboard(self, request):
# Rich data for web interface
return {
"user": await self._get_full_user_profile(request.user_id),
"orders": await self._get_detailed_orders(request.user_id),
"analytics": await self._get_user_analytics(request.user_id)
}When to Use API Gateway
- Microservices architecture: Multiple backend services need a unified entry point
- Cross-cutting concerns: Authentication, rate limiting, logging needed across services
- Response aggregation: Need to combine data from multiple services
- Protocol translation: External HTTP/REST to internal gRPC or message queues
- Client simplification: Hide backend complexity from clients
- Legacy integration: Modern API facade over legacy systems
Pros & Cons
✅ Pros
- Single entry point: Simplifies client code, one endpoint to call
- Cross-cutting concerns: Centralized auth, logging, rate limiting
- Response aggregation: Combine multiple service calls into one
- Protocol translation: External REST to internal gRPC/messaging
- Service evolution: Backend changes hidden from clients
- Security: Internal services not exposed directly to internet
❌ Cons
- Single point of failure: Gateway down means all services inaccessible
- Performance bottleneck: All traffic goes through one component
- Increased complexity: Another layer to develop and maintain
- Latency: Additional network hop for every request
- Coupling risk: Gateway logic can become tightly coupled to services
- Operational overhead: Need high availability, monitoring, scaling
💡 Pro Tip: Use established API Gateway solutions (Kong, AWS API Gateway, Azure APIM) instead of building from scratch. They handle authentication, rate limiting, caching, and monitoring out of the box.
Backend for Frontend (BFF)
Intent: Create separate backend services tailored to the needs of different frontend applications (web, mobile, desktop). Each BFF aggregates and transforms data from microservices to match its frontend's specific requirements.
💡 Problem It Solves: Mobile apps need different data than web apps (less data, optimized payloads). Forcing one API to serve all clients leads to over-fetching, under-fetching, and bloated responses.
Architecture
Without BFF: One API for All Clients
All clients share a single API Gateway - responses must satisfy every frontend
With BFF: Tailored Backend for Each Client
Each frontend gets its own BFF that fetches only what it needs from downstream services
Implementation Example
# Microservices return detailed data
class UserService:
def get_user(self, user_id):
return {
"id": user_id,
"name": "Alice",
"email": "alice@example.com",
"address": {...}, # Full address object
"preferences": {...}, # Many preferences
"metadata": {...} # Lots of metadata
}
class OrderService:
def get_orders(self, user_id):
return [{
"id": "order-1",
"items": [...], # Full item details
"shipping": {...},
"billing": {...},
"history": [...] # Complete order history
}]
# Web BFF - Optimized for web dashboard
class WebBFF:
def __init__(self, user_service, order_service, product_service):
self.user_service = user_service
self.order_service = order_service
self.product_service = product_service
def get_dashboard(self, user_id):
"""Aggregate data for web dashboard"""
# Fetch from multiple services in parallel
user = self.user_service.get_user(user_id)
orders = self.order_service.get_orders(user_id)
recommendations = self.product_service.get_recommendations(user_id)
# Transform to match web UI needs
return {
"user": {
"name": user["name"],
"email": user["email"],
"address": user["address"] # Web shows full address
},
"recentOrders": orders[:10], # Last 10 orders with full details
"recommendations": recommendations
}
# Mobile BFF - Optimized for mobile constraints
class MobileBFF:
def __init__(self, user_service, order_service, product_service):
self.user_service = user_service
self.order_service = order_service
self.product_service = product_service
def get_dashboard(self, user_id):
"""Aggregate data for mobile app - minimal payload"""
user = self.user_service.get_user(user_id)
orders = self.order_service.get_orders(user_id)
# Transform to minimize mobile bandwidth
return {
"user": {
"name": user["name"]
# No address - save bandwidth
},
"recentOrders": [
{
"id": o["id"],
"total": o["total"],
"status": o["status"]
# Minimal fields only
}
for o in orders[:5] # Only 5 orders on mobile
]
# No recommendations - mobile doesn't show them
}
# Desktop BFF - Power user features
class DesktopBFF:
def __init__(self, user_service, order_service, product_service):
self.user_service = user_service
self.order_service = order_service
self.product_service = product_service
def get_dashboard(self, user_id):
"""Full data for desktop power users"""
user = self.user_service.get_user(user_id)
orders = self.order_service.get_orders(user_id)
analytics = self.order_service.get_analytics(user_id)
return {
"user": user, # Everything
"allOrders": orders, # All orders
"analytics": analytics, # Advanced analytics
"exportData": self.prepare_export(orders) # Desktop-only feature
}
# Each frontend gets exactly what it needs!GraphQL as BFF
GraphQL naturally implements BFF pattern by allowing clients to request exactly the data they need:
# GraphQL BFF
type Query {
dashboard: Dashboard
}
type Dashboard {
user: User
recentOrders: [Order!]!
recommendations: [Product!]!
}
# Web client query - requests everything
query WebDashboard {
dashboard {
user {
name
email
address { street, city, zip }
}
recentOrders(limit: 10) {
id
total
items { name, quantity, price }
}
recommendations { id, name, image }
}
}
# Mobile client query - minimal fields
query MobileDashboard {
dashboard {
user { name }
recentOrders(limit: 5) {
id
total
status
}
}
}
# Each client gets exactly what it asks for!When to Use BFF
- Multiple client types: Web, mobile, smart TV, etc. with different needs
- Performance optimization: Mobile needs smaller payloads than web
- Different features: Desktop has features mobile doesn't
- Team structure: Separate teams for web and mobile can own their BFFs
- Rapid frontend iteration: Frontend teams can change BFF without affecting others
✅ Pros
- Optimized for each client type
- No over-fetching or under-fetching
- Frontend teams have autonomy
- Can evolve BFFs independently
- Better performance (tailored responses)
- Encapsulates backend complexity
❌ Cons
- Code duplication across BFFs
- More services to maintain
- Need to coordinate shared logic
- Risk of drift between BFFs
- Increased operational complexity
💡 Pro Tip: Start with one API (or GraphQL). Add BFFs only when you have clear evidence that different clients need significantly different data shapes or performance characteristics.
Clean Architecture
Intent: Organize code in concentric circles with dependencies pointing inward. Core business rules (entities) are at the center, surrounded by use cases, then interface adapters, then frameworks/drivers. Similar to hexagonal but with explicit layers.
# LAYER 1: ENTITIES (Core Business Rules)
class Order:
"""Enterprise business rules - innermost circle"""
def __init__(self, id, user_id, items):
self.id = id
self.user_id = user_id
self.items = items
self._status = "pending"
def calculate_total(self):
"""Business rule: How to calculate order total"""
return sum(item.price * item.quantity for item in self.items)
def can_be_cancelled(self):
"""Business rule: When can an order be cancelled"""
return self._status in ["pending", "confirmed"]
# LAYER 2: USE CASES (Application Business Rules)
class CreateOrderUseCase:
"""Application-specific business rules"""
def __init__(
self,
order_repo: OrderRepositoryPort,
payment_gateway: PaymentGatewayPort,
presenter: OrderPresenterPort
):
# Dependencies are abstractions (ports)
self.orders = order_repo
self.payment = payment_gateway
self.presenter = presenter
def execute(self, request: CreateOrderRequest):
"""Execute use case - orchestrates entities and ports"""
# Create entity
order = Order(
id=str(uuid.uuid4()),
user_id=request.user_id,
items=request.items
)
# Validate business rules
if order.calculate_total() <= 0:
return self.presenter.present_error("Invalid order total")
# Use port (abstraction) for external operation
if not self.payment.process(order.calculate_total(), order.user_id):
return self.presenter.present_error("Payment failed")
# Save via port
self.orders.save(order)
# Present result
return self.presenter.present_order(order)
# LAYER 3: INTERFACE ADAPTERS (Controllers, Gateways, Presenters)
class OrderRepositoryPort(ABC):
"""Port: Abstract interface"""
@abstractmethod
def save(self, order: Order): pass
@abstractmethod
def find_by_id(self, id: str) -> Order: pass
class PostgresOrderRepository(OrderRepositoryPort):
"""Adapter: Concrete implementation"""
def save(self, order: Order):
self.db.execute(
"INSERT INTO orders VALUES (%s, %s, %s)",
(order.id, order.user_id, order.calculate_total())
)
def find_by_id(self, id: str) -> Order:
row = self.db.query("SELECT * FROM orders WHERE id = %s", id)
return self._to_entity(row)
class OrderPresenterPort(ABC):
"""Port: How to present results"""
@abstractmethod
def present_order(self, order: Order): pass
@abstractmethod
def present_error(self, message: str): pass
class RESTOrderPresenter(OrderPresenterPort):
"""Adapter: REST API presentation"""
def present_order(self, order: Order):
return {
"order_id": order.id,
"total": order.calculate_total(),
"status": order._status
}
def present_error(self, message: str):
return {"error": message}, 400
class CLIOrderPresenter(OrderPresenterPort):
"""Adapter: Command-line presentation"""
def present_order(self, order: Order):
print(f"Order {order.id} created")
print(f"Total: {order.calculate_total()}")
def present_error(self, message: str):
print(f"Error: {message}")
# LAYER 4: FRAMEWORKS & DRIVERS (Web framework, DB, External services)
class FlaskOrderController:
"""Framework-specific controller"""
def __init__(self, create_order_use_case: CreateOrderUseCase):
self.create_order = create_order_use_case
@app.post("/orders")
def create_order_endpoint(self):
# Convert framework request to use case request
request = CreateOrderRequest(
user_id=flask.request.json["user_id"],
items=flask.request.json["items"]
)
# Execute use case
response = self.create_order.execute(request)
return response
# Dependency flow: Outer layers depend on inner layers
# Framework → Adapters → Use Cases → Entities
# Core business logic (entities, use cases) has no dependencies on frameworksWhen to Use Clean Architecture
- Complex business rules: Core domain logic is sophisticated and central to the application
- Long-lived applications: System will outlive frameworks and infrastructure
- Framework independence: Want flexibility to change frameworks without rewriting business logic
- High testability needs: Need to test business logic in complete isolation
- Multiple interfaces: Same business logic used by web, mobile, CLI, batch jobs
- Team organization: Different teams handle core vs infrastructure
Pros & Cons
✅ Pros
- Framework independence: Core logic not tied to any framework
- Testability: Business logic testable without UI, database, or external dependencies
- Database independence: Swap databases without changing business rules
- UI independence: Same logic works with any UI (web, mobile, CLI)
- Maintainability: Changes isolated to appropriate layers
- Business focus: Core domain models reflect business concepts
❌ Cons
- High initial complexity: Many layers, interfaces, and abstractions
- Boilerplate code: DTOs, mappers, adapters add significant code
- Learning curve: Difficult for developers unfamiliar with the pattern
- Over-engineering risk: Overkill for simple CRUD applications
- Development speed: Slower initial development than simpler architectures
- Indirection overhead: Multiple layers make code harder to trace
⚠️ Reality Check: Clean Architecture is excellent for complex, long-lived applications with sophisticated business logic. For simple CRUD apps or prototypes, it's over-engineering. Use when the benefits (testability, framework independence) justify the complexity cost.
Domain-Driven Design (DDD)
Intent: Structure software around the business domain. Focus on ubiquitous language, bounded contexts, aggregates, entities, and value objects. DDD is both strategic (how to organize large systems) and tactical (how to model domains).
Strategic DDD: Bounded Contexts
# Each bounded context has its own model of "Customer"
# SALES CONTEXT
class Customer:
"""Customer from sales perspective"""
def __init__(self, id, name, credit_limit):
self.id = id
self.name = name
self.credit_limit = credit_limit
self.orders = []
def can_place_order(self, amount):
"""Sales-specific business rule"""
total_pending = sum(o.total for o in self.orders if o.status == "pending")
return total_pending + amount <= self.credit_limit
# SHIPPING CONTEXT
class Customer:
"""Customer from shipping perspective - different model!"""
def __init__(self, id, shipping_address, preferred_carrier):
self.id = id
self.shipping_address = shipping_address
self.preferred_carrier = preferred_carrier
def get_shipping_options(self):
"""Shipping-specific logic"""
return [self.preferred_carrier, "standard", "express"]
# SUPPORT CONTEXT
class Customer:
"""Customer from support perspective"""
def __init__(self, id, name, support_tier, tickets):
self.id = id
self.name = name
self.support_tier = support_tier # "basic", "premium"
self.tickets = tickets
def can_open_ticket(self):
"""Support-specific rule"""
open_tickets = [t for t in self.tickets if t.status == "open"]
max_tickets = 10 if self.support_tier == "premium" else 3
return len(open_tickets) < max_tickets
# Contexts communicate via published events or APIs
# Not direct coupling - each context autonomousTactical DDD: Building Blocks
# 1. VALUE OBJECTS - Immutable, defined by attributes
class Money:
"""Value object: equality by value, immutable"""
def __init__(self, amount, currency):
self._amount = amount
self._currency = currency
@property
def amount(self):
return self._amount
def add(self, other):
if self._currency != other._currency:
raise ValueError("Currency mismatch")
return Money(self._amount + other._amount, self._currency)
def __eq__(self, other):
return (self._amount == other._amount and
self._currency == other._currency)
class Address:
"""Value object"""
def __init__(self, street, city, zip_code):
self.street = street
self.city = city
self.zip_code = zip_code
def __eq__(self, other):
return (self.street == other.street and
self.city == other.city and
self.zip_code == other.zip_code)
# 2. ENTITIES - Identity matters, mutable
class Order:
"""Entity: identity by ID, mutable state"""
def __init__(self, order_id):
self._id = order_id # Identity
self._items = []
self._status = "pending"
@property
def id(self):
return self._id
def add_item(self, product, quantity, price):
self._items.append(OrderItem(product, quantity, price))
def __eq__(self, other):
# Entities equal by ID, not attributes
return self._id == other._id
# 3. AGGREGATES - Consistency boundary
class Order: # Aggregate Root
"""Controls access to all entities in aggregate"""
def __init__(self, order_id, customer_id):
self._id = order_id
self._customer_id = customer_id
self._items = [] # Part of aggregate
self._shipping = None # Part of aggregate
def add_item(self, product_id, quantity):
"""Only way to add items - enforces invariants"""
if self._status == "shipped":
raise ValueError("Cannot modify shipped order")
item = OrderItem(product_id, quantity)
self._items.append(item)
def set_shipping_address(self, address: Address):
"""Control all changes through aggregate root"""
if not self._items:
raise ValueError("Cannot ship empty order")
self._shipping = ShippingInfo(address)
def calculate_total(self):
"""Aggregate ensures consistency"""
subtotal = sum(item.total for item in self._items)
shipping = self._shipping.cost if self._shipping else 0
return subtotal + shipping
class OrderItem: # Not accessible outside aggregate
"""Entity within aggregate - only Order can modify"""
def __init__(self, product_id, quantity):
self.product_id = product_id
self.quantity = quantity
# 4. REPOSITORIES - Persistence abstraction for aggregates
class OrderRepository(ABC):
"""Repository operates on aggregate roots only"""
@abstractmethod
def save(self, order: Order):
"""Save entire aggregate"""
pass
@abstractmethod
def find_by_id(self, order_id) -> Order:
"""Retrieve entire aggregate"""
pass
# No methods for OrderItem - access through Order
# 5. DOMAIN EVENTS - Significant occurrences
class OrderPlaced:
"""Domain event: something that happened"""
def __init__(self, order_id, customer_id, total):
self.order_id = order_id
self.customer_id = customer_id
self.total = total
self.occurred_at = datetime.now()
class Order:
def place_order(self):
if not self._items:
raise ValueError("Cannot place empty order")
self._status = "placed"
# Raise domain event
event = OrderPlaced(self._id, self._customer_id, self.calculate_total())
self._domain_events.append(event)
return event
# 6. DOMAIN SERVICES - Operations that don't belong to entity
class PricingService:
"""Domain service: crosses multiple aggregates"""
def calculate_order_price(
self,
order: Order,
customer: Customer,
promotions: List[Promotion]
):
base_price = order.calculate_total()
# Apply customer-specific discount
discount = customer.discount_rate
# Apply promotions
for promo in promotions:
if promo.applies_to(order):
discount += promo.discount_rate
return base_price * (1 - discount)When to Use Domain-Driven Design
- Complex business domain: Rich business logic and sophisticated domain rules
- Long-lived systems: Applications that will evolve over years
- Domain expert collaboration: Close partnership with business stakeholders
- Large systems: Multiple bounded contexts that need clear boundaries
- Ubiquitous language needed: Shared vocabulary between developers and business
- Strategic design: Need to manage complexity across team and organizational boundaries
Pros & Cons
✅ Pros
- Ubiquitous language: Shared vocabulary between business and developers
- Business focus: Code models business domain, not technical infrastructure
- Bounded contexts: Clear boundaries reduce complexity in large systems
- Rich domain models: Behavior and data together, not anemic models
- Maintainability: Changes aligned with business concepts
- Strategic design: Patterns for managing large-scale complexity
❌ Cons
- High learning curve: Difficult concepts, requires deep understanding
- Requires domain experts: Need close collaboration with business
- Time investment: Modeling takes time, slower initial development
- Over-engineering risk: Massive overkill for simple CRUD apps
- Team alignment needed: Entire team must understand DDD principles
- Complexity overhead: Many patterns and concepts to apply correctly
⚠️ Critical Warning: DDD is for complex business domains, not technical complexity. Don't use it for simple CRUD apps, data-centric systems, or when you don't have access to domain experts. The overhead is significant and only justified when modeling a genuinely complex business domain.
Pattern Relationships & Combinations
Architecture patterns don't exist in isolation. They often complement each other or serve as alternatives depending on your needs. Understanding these relationships helps you make better architectural decisions.
Patterns That Work Well Together
Microservices + Event-Driven + Saga
Microservices communicate via events, and sagas coordinate distributed transactions across services. This combination provides loose coupling with transaction consistency.
CQRS + Event Sourcing
Event Sourcing naturally separates writes (events) from reads (projections). CQRS formalizes this separation with different models for commands and queries.
Hexagonal + DDD
Hexagonal Architecture protects domain logic from infrastructure. DDD provides patterns for modeling that domain logic. Together they create maintainable, business-focused systems.
Microservices + API Gateway + BFF
API Gateway provides single entry point to microservices. BFF pattern creates specialized gateways for different frontends (web, mobile) with optimized data aggregation.
Clean Architecture + DDD
Clean Architecture defines the layers and dependency rules. DDD provides tactical patterns (entities, value objects, aggregates) for the domain layer. Perfect combination for complex business logic.
Microservices + Database per Service + Event-Driven
Each microservice owns its database. Services stay synchronized through events, avoiding direct database access and maintaining loose coupling.
Alternative Patterns (Choose One)
Hexagonal vs Clean vs Layered
All three organize code architecture
- Layered: Simplest, horizontal layers, database-centric
- Hexagonal: Ports & Adapters, protects core from infrastructure
- Clean: Concentric circles, strictest dependency rules
Monolithic vs Microservices
Deployment architecture decision
- Monolithic: Single deployable, simpler operations
- Microservices: Independently deployable services
- Modular Monolith: Middle ground with internal boundaries
API Gateway vs BFF
Frontend integration patterns
- API Gateway: Single gateway for all clients
- BFF: Separate backend for each frontend type
- BFF is specialized version of API Gateway pattern
Event-Driven vs Choreography vs Orchestration
Service coordination approaches
- Event-Driven: Services react to events independently
- Saga Choreography: Distributed event-based coordination
- Saga Orchestration: Central coordinator manages workflow
Common Real-World Combinations
# E-Commerce Platform (Large Scale) Architecture: Deployment: Microservices Communication: Event-Driven + Saga (for orders) Frontend: API Gateway with BFFs (Web/Mobile) Data: Database per Service + Event Sourcing (orders) Individual Services: Clean Architecture + DDD # Banking System (Complex Domain) Architecture: Deployment: Modular Monolith (evolving to microservices) Internal Structure: Clean Architecture + Hexagonal Domain Modeling: Domain-Driven Design Read Optimization: CQRS for reporting Audit: Event Sourcing for transactions # SaaS Application (Medium Scale) Architecture: Deployment: Monolithic with modules Structure: Layered or Hexagonal Architecture Communication: Synchronous APIs Scaling: Read replicas, caching Evolution Path: Extract microservices as needed # Real-Time Analytics Platform Architecture: Deployment: Microservices Communication: Event-Driven Architecture Query: CQRS (Event Sourcing for writes, optimized read models) Frontend: BFF for dashboard vs API clients Data Flow: Stream processing + Event Store
💡 Golden Rule: Don't use all patterns at once. Start simple and add patterns only when they solve specific problems you're facing. Each pattern adds complexity that must be justified by concrete benefits.
Choosing the Right Architecture
| Pattern | Best For | Avoid When |
|---|---|---|
| Monolithic | Small teams, MVPs, simple domains, startups | Multiple teams, need independent scaling |
| Microservices | Large teams, independent scaling, polyglot | Small teams, simple domains, tight coupling |
| Event-Driven | Loose coupling, async workflows, scalability | Strong consistency required, simple flows |
| CQRS | Read/write scale differently, complex queries | Simple CRUD, small scale |
| Hexagonal | Testability, framework independence | Simple apps, rapid prototyping |
| Layered | Traditional apps, clear separation | Complex domains, need flexibility |
| Clean Architecture | Long-lived systems, testability critical | Throwaway prototypes, simple tools |
| DDD | Complex business logic, expert collaboration | Simple CRUD, data-centric apps |
Real-World Considerations
Start Simple, Evolve Deliberately
# Evolution path for a typical system
# Phase 1: MVP - Monolithic with layers
monolith = LayeredMonolith()
# Simple, fast to build, easy to deploy
# Phase 2: Growing - Extract high-load services
system = {
"monolith": CoreMonolith(),
"image_service": ImageMicroservice(), # Heavy processing
"email_service": EmailMicroservice() # High volume
}
# Phase 3: Scale - More microservices, event-driven
system = {
"api_gateway": APIGateway(),
"user_service": UserMicroservice(),
"order_service": OrderMicroservice(),
"inventory_service": InventoryMicroservice(),
"event_bus": EventBus()
}
# Phase 4: Optimization - CQRS for read-heavy services
system = {
"api_gateway": APIGateway(),
"user_write": UserCommandService(),
"user_read": UserQueryService(), # Separate read model
"order_service": OrderCQRS() # CQRS for complex queries
}
# Don't jump to Phase 4 on day 1!Hybrid Approaches
- Modular Monolith: Monolithic deployment with microservice-like boundaries internally
- Mini-services: Somewhere between monolith and full microservices
- Selective CQRS: Apply CQRS only to services that need it
- Partial Event Sourcing: Event sourcing for critical domains, traditional for others
Common Mistakes
❌ Resume-Driven Development
Choosing microservices/CQRS/event sourcing because they're trendy, not because they solve real problems. These patterns add complexity, make sure the benefits justify the cost.
❌ Distributed Monolith
Creating microservices that call each other synchronously with tight coupling. You get all the complexity of distributed systems with none of the benefits. This is worse than a monolith.
❌ Premature Decomposition
Splitting into microservices before understanding the domain boundaries. Get it wrong and you'll spend years paying the price. Start with a monolith, learn the domain, then extract services at natural boundaries.
Key Takeaways
- Architecture patterns solve system-level problems like scalability, maintainability, team autonomy
- Monoliths are underrated: Start here unless you have specific reasons not to
- Microservices trade complexity for flexibility: Only adopt when benefits outweigh costs
- Event-driven architecture enables loose coupling but introduces eventual consistency
- CQRS separates reads and writes for independent optimization
- Hexagonal/Clean Architecture protects business logic from external dependencies
- DDD helps model complex domains through bounded contexts and tactical patterns
- Sagas manage distributed transactions across services
- No silver bullet: Every architecture pattern is a trade-off
- Context matters: Choose based on team size, domain complexity, scale requirements, and constraints
- Evolution over revolution: Start simple, evolve architecture as needs change