📘 Design Patterns: Advanced Patterns

🎯 Introduction

Welcome to this exciting deep dive into advanced design patterns in Python! 🎉 In this guide, we’ll explore sophisticated patterns that go beyond the classic Gang of Four patterns, taking your Python skills to the next level.

You’ll discover how advanced design patterns can transform your code architecture, making it more flexible, maintainable, and elegant. Whether you’re building microservices 🌐, event-driven systems 🔄, or complex applications 🏗️, mastering these patterns is essential for writing professional-grade Python code.

By the end of this tutorial, you’ll feel confident implementing advanced patterns like the Repository pattern, CQRS, Event Sourcing, and more in your own projects! Let’s dive in! 🏊‍♂️

📚 Understanding Advanced Design Patterns

🤔 What are Advanced Design Patterns?

Advanced design patterns are like sophisticated blueprints for solving complex architectural problems 🎨. Think of them as specialized tools in a master craftsman’s workshop - each designed for specific, intricate tasks that basic tools can’t handle elegantly.

In Python terms, these patterns help you build systems that can scale, adapt to change, and handle complex business logic with grace. This means you can:

• ✨ Build systems that scale effortlessly
• 🚀 Separate concerns for cleaner architecture
• 🛡️ Handle complex domain logic elegantly
• 🔄 Create event-driven architectures
• 📊 Implement sophisticated data access patterns

💡 Why Use Advanced Patterns?

Here’s why senior developers love advanced patterns:

1. Scalability 🚀: Design systems that grow with your needs
2. Maintainability 🔧: Keep complex systems manageable
3. Testability 🧪: Write unit tests with ease
4. Flexibility 🎯: Adapt to changing requirements
5. Performance ⚡: Optimize critical paths effectively

Real-world example: Imagine building an e-commerce platform 🛒. With advanced patterns like Repository and CQRS, you can separate your read and write operations, implement caching strategies, and handle millions of transactions efficiently!

🔧 Basic Syntax and Usage

📝 Repository Pattern

Let’s start with the Repository pattern - your data access superhero:

# 👋 Hello, Repository Pattern!
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any
import json

# 🎨 Define our domain model
class Product:
    def __init__(self, id: str, name: str, price: float, stock: int):
        self.id = id
        self.name = name
        self.price = price
        self.stock = stock
        self.emoji = "📦"  # Every product needs an emoji! 
    
    def __repr__(self):
        return f"{self.emoji} Product({self.name}, ${self.price})"

# 🛡️ Abstract repository interface
class Repository(ABC):
    @abstractmethod
    def add(self, entity: Any) -> None:
        pass
    
    @abstractmethod
    def get(self, id: str) -> Optional[Any]:
        pass
    
    @abstractmethod
    def get_all(self) -> List[Any]:
        pass
    
    @abstractmethod
    def update(self, entity: Any) -> None:
        pass
    
    @abstractmethod
    def delete(self, id: str) -> None:
        pass

# 🚀 Concrete implementation
class ProductRepository(Repository):
    def __init__(self):
        self._storage: Dict[str, Product] = {}
        print("🏪 Product repository initialized!")
    
    def add(self, product: Product) -> None:
        self._storage[product.id] = product
        print(f"✅ Added {product.emoji} {product.name} to repository!")
    
    def get(self, id: str) -> Optional[Product]:
        return self._storage.get(id)
    
    def get_all(self) -> List[Product]:
        return list(self._storage.values())
    
    def update(self, product: Product) -> None:
        if product.id in self._storage:
            self._storage[product.id] = product
            print(f"📝 Updated {product.emoji} {product.name}")
    
    def delete(self, id: str) -> None:
        if id in self._storage:
            product = self._storage.pop(id)
            print(f"🗑️ Deleted {product.emoji} {product.name}")

💡 Explanation: The Repository pattern abstracts data access, making it easy to switch between different storage mechanisms (database, file, memory) without changing your business logic!

🎯 Unit of Work Pattern

Now let’s combine it with Unit of Work for transaction management:

# 🏗️ Unit of Work pattern for transaction management
class UnitOfWork:
    def __init__(self):
        self._products = ProductRepository()
        self._new: List[Product] = []
        self._dirty: List[Product] = []
        self._removed: List[str] = []
        print("💼 Unit of Work initialized!")
    
    # 🎨 Track new entities
    def register_new(self, product: Product):
        self._new.append(product)
        print(f"📝 Registered new: {product.emoji} {product.name}")
    
    # 🔄 Track modified entities
    def register_dirty(self, product: Product):
        if product not in self._dirty:
            self._dirty.append(product)
            print(f"✏️ Registered dirty: {product.emoji} {product.name}")
    
    # 🗑️ Track removed entities
    def register_removed(self, id: str):
        self._removed.append(id)
        print(f"🗑️ Registered for removal: {id}")
    
    # 💾 Commit all changes
    def commit(self):
        print("🚀 Committing changes...")
        
        # Add new products
        for product in self._new:
            self._products.add(product)
        
        # Update modified products
        for product in self._dirty:
            self._products.update(product)
        
        # Remove deleted products
        for id in self._removed:
            self._products.delete(id)
        
        # Clear tracking lists
        self._new.clear()
        self._dirty.clear()
        self._removed.clear()
        
        print("✅ All changes committed!")
    
    # 🔄 Rollback changes
    def rollback(self):
        self._new.clear()
        self._dirty.clear()
        self._removed.clear()
        print("↩️ Changes rolled back!")

💡 Practical Examples

🛒 Example 1: E-commerce with CQRS Pattern

Let’s build a CQRS (Command Query Responsibility Segregation) system:

# 🛍️ CQRS Pattern - Separate read and write models
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict
import uuid

# 📝 Commands (write operations)
@dataclass
class Command:
    timestamp: datetime = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()

@dataclass
class CreateOrderCommand(Command):
    customer_id: str
    items: List[Dict[str, Any]]
    emoji: str = "🛒"

@dataclass
class UpdateOrderStatusCommand(Command):
    order_id: str
    status: str
    emoji: str = "📦"

# 🔍 Queries (read operations)
@dataclass
class Query:
    pass

@dataclass
class GetOrderByIdQuery(Query):
    order_id: str

@dataclass
class GetOrdersByCustomerQuery(Query):
    customer_id: str

# 🎯 Command handlers
class CommandHandler:
    def __init__(self):
        self.events = []  # Event store
        print("⚡ Command handler ready!")
    
    def handle(self, command: Command):
        if isinstance(command, CreateOrderCommand):
            order_id = str(uuid.uuid4())
            event = {
                'type': 'OrderCreated',
                'order_id': order_id,
                'customer_id': command.customer_id,
                'items': command.items,
                'timestamp': command.timestamp,
                'emoji': command.emoji
            }
            self.events.append(event)
            print(f"{command.emoji} Order {order_id} created!")
            return order_id
        
        elif isinstance(command, UpdateOrderStatusCommand):
            event = {
                'type': 'OrderStatusUpdated',
                'order_id': command.order_id,
                'status': command.status,
                'timestamp': command.timestamp,
                'emoji': command.emoji
            }
            self.events.append(event)
            print(f"{command.emoji} Order {command.order_id} status: {command.status}")

# 📊 Query handlers with read model
class QueryHandler:
    def __init__(self, events):
        self.events = events
        self._orders = {}  # Read model
        self._rebuild_read_model()
        print("🔍 Query handler ready!")
    
    def _rebuild_read_model(self):
        # Rebuild read model from events
        for event in self.events:
            if event['type'] == 'OrderCreated':
                self._orders[event['order_id']] = {
                    'id': event['order_id'],
                    'customer_id': event['customer_id'],
                    'items': event['items'],
                    'status': 'created',
                    'created_at': event['timestamp']
                }
            elif event['type'] == 'OrderStatusUpdated':
                if event['order_id'] in self._orders:
                    self._orders[event['order_id']]['status'] = event['status']
    
    def handle(self, query: Query):
        if isinstance(query, GetOrderByIdQuery):
            return self._orders.get(query.order_id)
        
        elif isinstance(query, GetOrdersByCustomerQuery):
            return [
                order for order in self._orders.values()
                if order['customer_id'] == query.customer_id
            ]

# 🎮 Let's use CQRS!
command_handler = CommandHandler()
query_handler = QueryHandler(command_handler.events)

# Create an order
create_cmd = CreateOrderCommand(
    customer_id="cust-123",
    items=[
        {"name": "Python Book", "price": 29.99, "emoji": "📘"},
        {"name": "Coffee", "price": 4.99, "emoji": "☕"}
    ]
)
order_id = command_handler.handle(create_cmd)

# Update order status
update_cmd = UpdateOrderStatusCommand(order_id=order_id, status="shipped")
command_handler.handle(update_cmd)

# Query the order
query_handler = QueryHandler(command_handler.events)  # Rebuild read model
order = query_handler.handle(GetOrderByIdQuery(order_id=order_id))
print(f"📋 Order details: {order}")

🎯 Try it yourself: Add a cancel order command and implement event replay functionality!

🎮 Example 2: Event Sourcing Pattern

Let’s implement event sourcing for game state:

# 🏆 Event Sourcing for a game
from abc import ABC, abstractmethod
from datetime import datetime
from typing import List, Dict, Any
import json

# 🎯 Base event class
class Event(ABC):
    def __init__(self, aggregate_id: str):
        self.aggregate_id = aggregate_id
        self.timestamp = datetime.now()
        self.version = 1
    
    @abstractmethod
    def apply(self, state: Dict[str, Any]) -> Dict[str, Any]:
        pass

# 🎮 Game events
class GameStartedEvent(Event):
    def __init__(self, player_id: str, player_name: str):
        super().__init__(player_id)
        self.player_name = player_name
        self.emoji = "🎮"
    
    def apply(self, state: Dict[str, Any]) -> Dict[str, Any]:
        return {
            **state,
            'player_id': self.aggregate_id,
            'player_name': self.player_name,
            'score': 0,
            'level': 1,
            'achievements': ["🌟 First Steps"],
            'status': 'playing'
        }

class PointsEarnedEvent(Event):
    def __init__(self, player_id: str, points: int):
        super().__init__(player_id)
        self.points = points
        self.emoji = "✨"
    
    def apply(self, state: Dict[str, Any]) -> Dict[str, Any]:
        new_score = state['score'] + self.points
        new_level = (new_score // 100) + 1
        
        new_state = {
            **state,
            'score': new_score,
            'level': new_level
        }
        
        # 🎊 Level up achievement
        if new_level > state['level']:
            new_state['achievements'] = state['achievements'] + [f"🏆 Level {new_level} Master"]
        
        return new_state

class PowerUpCollectedEvent(Event):
    def __init__(self, player_id: str, power_up: str):
        super().__init__(player_id)
        self.power_up = power_up
        self.emoji = "💪"
    
    def apply(self, state: Dict[str, Any]) -> Dict[str, Any]:
        power_ups = state.get('power_ups', [])
        return {
            **state,
            'power_ups': power_ups + [self.power_up]
        }

# 📚 Event store
class EventStore:
    def __init__(self):
        self._events: Dict[str, List[Event]] = {}
        print("💾 Event store initialized!")
    
    def save_event(self, event: Event):
        if event.aggregate_id not in self._events:
            self._events[event.aggregate_id] = []
        
        # Set version number
        event.version = len(self._events[event.aggregate_id]) + 1
        self._events[event.aggregate_id].append(event)
        
        print(f"{getattr(event, 'emoji', '📝')} Event saved: {event.__class__.__name__} v{event.version}")
    
    def get_events(self, aggregate_id: str) -> List[Event]:
        return self._events.get(aggregate_id, [])

# 🎮 Game aggregate
class GamePlayer:
    def __init__(self, event_store: EventStore):
        self.event_store = event_store
        self._state: Dict[str, Any] = {}
    
    def start_game(self, player_id: str, player_name: str):
        event = GameStartedEvent(player_id, player_name)
        self.event_store.save_event(event)
        print(f"🎮 {player_name} started playing!")
    
    def earn_points(self, player_id: str, points: int):
        event = PointsEarnedEvent(player_id, points)
        self.event_store.save_event(event)
        
        # Get current state to show level up
        state = self.get_player_state(player_id)
        print(f"✨ Earned {points} points! Score: {state['score']}")
    
    def collect_power_up(self, player_id: str, power_up: str):
        event = PowerUpCollectedEvent(player_id, power_up)
        self.event_store.save_event(event)
        print(f"💪 Collected power-up: {power_up}")
    
    def get_player_state(self, player_id: str) -> Dict[str, Any]:
        # Rebuild state from events
        events = self.event_store.get_events(player_id)
        state = {}
        
        for event in events:
            state = event.apply(state)
        
        return state
    
    def replay_game_history(self, player_id: str):
        print(f"\n🎬 Replaying game history for player {player_id}:")
        events = self.event_store.get_events(player_id)
        
        state = {}
        for i, event in enumerate(events):
            state = event.apply(state)
            print(f"  📹 Event {i+1}: {event.__class__.__name__} - Score: {state.get('score', 0)}")

# 🎮 Let's play!
event_store = EventStore()
game = GamePlayer(event_store)

# Start a new game
game.start_game("player-1", "Alice")

# Play the game
game.earn_points("player-1", 50)
game.collect_power_up("player-1", "🚀 Speed Boost")
game.earn_points("player-1", 75)
game.collect_power_up("player-1", "🛡️ Shield")
game.earn_points("player-1", 100)

# Check final state
final_state = game.get_player_state("player-1")
print(f"\n📊 Final state: Level {final_state['level']}, Score: {final_state['score']}")
print(f"🏆 Achievements: {', '.join(final_state['achievements'])}")
print(f"💪 Power-ups: {', '.join(final_state.get('power_ups', []))}")

# Replay history
game.replay_game_history("player-1")

🚀 Advanced Concepts

🧙‍♂️ Specification Pattern

When you’re ready to level up, try the Specification pattern for complex business rules:

# 🎯 Specification pattern for complex business rules
from abc import ABC, abstractmethod
from typing import Any, List

class Specification(ABC):
    @abstractmethod
    def is_satisfied_by(self, candidate: Any) -> bool:
        pass
    
    def and_(self, other: 'Specification') -> 'AndSpecification':
        return AndSpecification(self, other)
    
    def or_(self, other: 'Specification') -> 'OrSpecification':
        return OrSpecification(self, other)
    
    def not_(self) -> 'NotSpecification':
        return NotSpecification(self)

# 🔗 Composite specifications
class AndSpecification(Specification):
    def __init__(self, left: Specification, right: Specification):
        self.left = left
        self.right = right
    
    def is_satisfied_by(self, candidate: Any) -> bool:
        return self.left.is_satisfied_by(candidate) and self.right.is_satisfied_by(candidate)

class OrSpecification(Specification):
    def __init__(self, left: Specification, right: Specification):
        self.left = left
        self.right = right
    
    def is_satisfied_by(self, candidate: Any) -> bool:
        return self.left.is_satisfied_by(candidate) or self.right.is_satisfied_by(candidate)

class NotSpecification(Specification):
    def __init__(self, spec: Specification):
        self.spec = spec
    
    def is_satisfied_by(self, candidate: Any) -> bool:
        return not self.spec.is_satisfied_by(candidate)

# 🛍️ Business rule specifications
class PremiumCustomerSpec(Specification):
    def is_satisfied_by(self, customer: Dict[str, Any]) -> bool:
        return customer.get('tier') == 'premium'

class HighValueOrderSpec(Specification):
    def __init__(self, threshold: float):
        self.threshold = threshold
    
    def is_satisfied_by(self, order: Dict[str, Any]) -> bool:
        return order.get('total', 0) >= self.threshold

class RecentCustomerSpec(Specification):
    def __init__(self, days: int):
        self.days = days
    
    def is_satisfied_by(self, customer: Dict[str, Any]) -> bool:
        # Simplified: check if registered within days
        return customer.get('days_since_registration', float('inf')) <= self.days

# 🎨 Using specifications
premium_spec = PremiumCustomerSpec()
high_value_spec = HighValueOrderSpec(100.0)
recent_spec = RecentCustomerSpec(30)

# Complex business rule: Premium OR (High value AND Recent)
eligible_for_discount = premium_spec.or_(high_value_spec.and_(recent_spec))

# Test customers
customers = [
    {"name": "Alice", "tier": "premium", "order_total": 50, "days_since_registration": 60},
    {"name": "Bob", "tier": "regular", "order_total": 150, "days_since_registration": 15},
    {"name": "Charlie", "tier": "regular", "order_total": 30, "days_since_registration": 5},
]

for customer in customers:
    order = {"total": customer["order_total"]}
    if eligible_for_discount.is_satisfied_by(customer) and high_value_spec.is_satisfied_by(order):
        print(f"✅ {customer['name']} is eligible for discount! 🎉")
    else:
        print(f"❌ {customer['name']} is not eligible")

🏗️ Saga Pattern for Distributed Transactions

For the brave developers - implement distributed transactions:

# 🚀 Saga pattern for distributed transactions
from enum import Enum
from typing import List, Callable, Any, Optional
import uuid

class SagaStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"

class SagaStep:
    def __init__(self, name: str, action: Callable, compensation: Callable):
        self.name = name
        self.action = action
        self.compensation = compensation
        self.status = SagaStatus.PENDING
        self.result = None
        self.error = None
        self.emoji = "📝"
    
    def execute(self) -> bool:
        try:
            print(f"▶️ Executing: {self.name}")
            self.status = SagaStatus.RUNNING
            self.result = self.action()
            self.status = SagaStatus.COMPLETED
            print(f"✅ Completed: {self.name}")
            return True
        except Exception as e:
            self.status = SagaStatus.FAILED
            self.error = str(e)
            print(f"❌ Failed: {self.name} - {e}")
            return False
    
    def compensate(self) -> bool:
        if self.status != SagaStatus.COMPLETED:
            return True  # Nothing to compensate
        
        try:
            print(f"↩️ Compensating: {self.name}")
            self.status = SagaStatus.COMPENSATING
            self.compensation(self.result)
            print(f"✅ Compensated: {self.name}")
            return True
        except Exception as e:
            print(f"⚠️ Compensation failed: {self.name} - {e}")
            return False

class Saga:
    def __init__(self, name: str):
        self.id = str(uuid.uuid4())
        self.name = name
        self.steps: List[SagaStep] = []
        self.status = SagaStatus.PENDING
        print(f"📋 Saga '{name}' created with ID: {self.id}")
    
    def add_step(self, step: SagaStep):
        self.steps.append(step)
        print(f"➕ Added step: {step.name}")
    
    def execute(self):
        print(f"\n🚀 Starting saga: {self.name}")
        self.status = SagaStatus.RUNNING
        
        completed_steps = []
        
        for step in self.steps:
            if step.execute():
                completed_steps.append(step)
            else:
                # Step failed, compensate completed steps
                print(f"\n⚠️ Saga failed at step: {step.name}")
                self.status = SagaStatus.COMPENSATING
                
                # Compensate in reverse order
                for completed_step in reversed(completed_steps):
                    completed_step.compensate()
                
                self.status = SagaStatus.FAILED
                print(f"❌ Saga '{self.name}' failed and compensated")
                return False
        
        self.status = SagaStatus.COMPLETED
        print(f"\n🎉 Saga '{self.name}' completed successfully!")
        return True

# 🛒 Example: Order processing saga
class OrderService:
    def __init__(self):
        self.orders = {}
    
    def create_order(self, order_id: str, items: List[str]) -> str:
        self.orders[order_id] = {"id": order_id, "items": items, "status": "created"}
        print(f"  📦 Order {order_id} created")
        return order_id
    
    def cancel_order(self, order_id: str):
        if order_id in self.orders:
            self.orders[order_id]["status"] = "cancelled"
            print(f"  🗑️ Order {order_id} cancelled")

class PaymentService:
    def __init__(self):
        self.payments = {}
        self.balance = 1000  # Starting balance
    
    def process_payment(self, payment_id: str, amount: float) -> str:
        if amount > self.balance:
            raise Exception("Insufficient funds! 💸")
        
        self.balance -= amount
        self.payments[payment_id] = {"id": payment_id, "amount": amount, "status": "processed"}
        print(f"  💳 Payment {payment_id} processed: ${amount}")
        return payment_id
    
    def refund_payment(self, payment_id: str):
        if payment_id in self.payments:
            amount = self.payments[payment_id]["amount"]
            self.balance += amount
            self.payments[payment_id]["status"] = "refunded"
            print(f"  💰 Payment {payment_id} refunded: ${amount}")

class InventoryService:
    def __init__(self):
        self.inventory = {"item1": 10, "item2": 5, "item3": 0}
    
    def reserve_items(self, reservation_id: str, items: List[str]) -> str:
        for item in items:
            if self.inventory.get(item, 0) <= 0:
                raise Exception(f"Item {item} out of stock! 📦")
            
            self.inventory[item] -= 1
        
        print(f"  📋 Items reserved: {items}")
        return reservation_id
    
    def release_items(self, reservation_id: str):
        # Simplified: just add items back
        print(f"  📤 Items released for reservation {reservation_id}")

# 🎮 Create and execute saga
order_service = OrderService()
payment_service = PaymentService()
inventory_service = InventoryService()

# Create order processing saga
saga = Saga("Process Online Order")

# Add saga steps
saga.add_step(SagaStep(
    "Create Order",
    lambda: order_service.create_order("order-123", ["item1", "item2"]),
    lambda order_id: order_service.cancel_order(order_id)
))

saga.add_step(SagaStep(
    "Reserve Inventory",
    lambda: inventory_service.reserve_items("reservation-123", ["item1", "item2"]),
    lambda reservation_id: inventory_service.release_items(reservation_id)
))

saga.add_step(SagaStep(
    "Process Payment",
    lambda: payment_service.process_payment("payment-123", 99.99),
    lambda payment_id: payment_service.refund_payment(payment_id)
))

# Execute the saga
saga.execute()

# Try a failing saga
print("\n" + "="*50 + "\n")

failing_saga = Saga("Process Large Order")

failing_saga.add_step(SagaStep(
    "Create Large Order",
    lambda: order_service.create_order("order-456", ["item3"]),  # Out of stock!
    lambda order_id: order_service.cancel_order(order_id)
))

failing_saga.add_step(SagaStep(
    "Reserve Out-of-Stock Items",
    lambda: inventory_service.reserve_items("reservation-456", ["item3"]),  # Will fail!
    lambda reservation_id: inventory_service.release_items(reservation_id)
))

failing_saga.add_step(SagaStep(
    "Process Large Payment",
    lambda: payment_service.process_payment("payment-456", 5000),  # Too expensive!
    lambda payment_id: payment_service.refund_payment(payment_id)
))

failing_saga.execute()

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Over-Engineering

# ❌ Wrong way - too complex for simple needs!
class OverEngineeredUserService:
    def __init__(self, 
                 repository_factory,
                 unit_of_work_factory,
                 event_store,
                 command_bus,
                 query_bus,
                 saga_orchestrator,
                 specification_engine):
        # 😰 This is overkill for basic CRUD!
        pass

# ✅ Correct way - start simple!
class SimpleUserService:
    def __init__(self, db_connection):
        self.db = db_connection
        print("✨ Simple and effective!")
    
    def create_user(self, name: str, email: str):
        # Direct and clean
        return self.db.insert("users", {"name": name, "email": email})

🤯 Pitfall 2: Ignoring Transaction Boundaries

# ❌ Dangerous - no transaction management!
def transfer_money_unsafe(from_account: str, to_account: str, amount: float):
    # What if this fails after debit? 💥
    debit_account(from_account, amount)
    credit_account(to_account, amount)  # System crash here = money lost!

# ✅ Safe - use Unit of Work or Saga!
def transfer_money_safe(uow: UnitOfWork, from_account: str, to_account: str, amount: float):
    try:
        uow.accounts.debit(from_account, amount)
        uow.accounts.credit(to_account, amount)
        uow.commit()  # ✅ All or nothing!
        print("💰 Transfer successful!")
    except Exception as e:
        uow.rollback()  # 🛡️ Safe rollback
        print(f"⚠️ Transfer failed: {e}")

🛠️ Best Practices

1. 🎯 Choose Wisely: Not every problem needs an advanced pattern!
2. 📝 Document Intent: Explain WHY you chose a pattern
3. 🧪 Test Thoroughly: Advanced patterns need comprehensive tests
4. 🎨 Keep It Pythonic: Don’t fight the language
5. ✨ Iterate: Start simple, evolve as needed
6. 🔍 Monitor: Track pattern effectiveness in production

🧪 Hands-On Exercise

🎯 Challenge: Build a Microservices Communication System

Create a complete system using multiple advanced patterns:

📋 Requirements:

• ✅ Implement message bus for service communication
• 🏷️ Use Repository pattern for data access
• 👤 Apply CQRS for read/write separation
• 📅 Add event sourcing for audit trail
• 🎨 Include saga for distributed transactions
• 🛡️ Use specification pattern for business rules

🚀 Bonus Points:

• Add circuit breaker pattern for resilience
• Implement bulkhead pattern for isolation
• Create a service mesh simulation

💡 Solution

# 🎯 Microservices communication system!
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Callable, Any, Optional
from enum import Enum
import uuid
import time

# 📨 Message bus for inter-service communication
class Message(ABC):
    def __init__(self):
        self.id = str(uuid.uuid4())
        self.timestamp = datetime.now()
        self.emoji = "📨"

@dataclass
class OrderPlacedMessage(Message):
    order_id: str
    customer_id: str
    total: float
    
    def __init__(self, order_id: str, customer_id: str, total: float):
        super().__init__()
        self.order_id = order_id
        self.customer_id = customer_id
        self.total = total
        self.emoji = "🛒"

class MessageBus:
    def __init__(self):
        self._handlers: Dict[type, List[Callable]] = {}
        print("🚌 Message bus initialized!")
    
    def subscribe(self, message_type: type, handler: Callable):
        if message_type not in self._handlers:
            self._handlers[message_type] = []
        self._handlers[message_type].append(handler)
        print(f"📌 Subscribed to {message_type.__name__}")
    
    def publish(self, message: Message):
        message_type = type(message)
        handlers = self._handlers.get(message_type, [])
        
        print(f"{message.emoji} Publishing {message_type.__name__}")
        for handler in handlers:
            try:
                handler(message)
            except Exception as e:
                print(f"⚠️ Handler error: {e}")

# 🏗️ Service base with repository and CQRS
class ServiceBase:
    def __init__(self, name: str, message_bus: MessageBus):
        self.name = name
        self.message_bus = message_bus
        self.event_store = []
        self.read_model = {}
        print(f"🏢 {name} service initialized!")
    
    def record_event(self, event: Dict[str, Any]):
        event['service'] = self.name
        event['timestamp'] = datetime.now()
        self.event_store.append(event)
        self._update_read_model(event)
    
    def _update_read_model(self, event: Dict[str, Any]):
        # Override in subclasses
        pass

# 🛍️ Order service
class OrderService(ServiceBase):
    def __init__(self, message_bus: MessageBus):
        super().__init__("OrderService", message_bus)
        self.orders = {}  # Repository
    
    def place_order(self, customer_id: str, items: List[Dict[str, Any]]) -> str:
        order_id = f"order-{uuid.uuid4().hex[:8]}"
        total = sum(item['price'] * item['quantity'] for item in items)
        
        # Command: Create order
        order = {
            'id': order_id,
            'customer_id': customer_id,
            'items': items,
            'total': total,
            'status': 'pending'
        }
        self.orders[order_id] = order
        
        # Record event
        self.record_event({
            'type': 'OrderPlaced',
            'order': order
        })
        
        # Publish message
        self.message_bus.publish(
            OrderPlacedMessage(order_id, customer_id, total)
        )
        
        print(f"📦 Order {order_id} placed! Total: ${total}")
        return order_id
    
    def _update_read_model(self, event: Dict[str, Any]):
        if event['type'] == 'OrderPlaced':
            order = event['order']
            self.read_model[order['id']] = {
                'id': order['id'],
                'customer_id': order['customer_id'],
                'total': order['total'],
                'item_count': len(order['items'])
            }

# 💳 Payment service with saga
class PaymentService(ServiceBase):
    def __init__(self, message_bus: MessageBus):
        super().__init__("PaymentService", message_bus)
        self.payments = {}
        self.customer_balances = {
            'cust-123': 500.0,
            'cust-456': 100.0
        }
        
        # Subscribe to order events
        message_bus.subscribe(OrderPlacedMessage, self.handle_order_placed)
    
    def handle_order_placed(self, message: OrderPlacedMessage):
        print(f"💳 Processing payment for order {message.order_id}")
        
        # Check if customer can afford it
        balance = self.customer_balances.get(message.customer_id, 0)
        
        if balance >= message.total:
            # Process payment
            payment_id = f"pay-{uuid.uuid4().hex[:8]}"
            self.payments[payment_id] = {
                'id': payment_id,
                'order_id': message.order_id,
                'amount': message.total,
                'status': 'completed'
            }
            self.customer_balances[message.customer_id] -= message.total
            
            self.record_event({
                'type': 'PaymentCompleted',
                'payment_id': payment_id,
                'order_id': message.order_id,
                'amount': message.total
            })
            
            print(f"✅ Payment {payment_id} completed!")
        else:
            print(f"❌ Insufficient balance for customer {message.customer_id}")
            
            self.record_event({
                'type': 'PaymentFailed',
                'order_id': message.order_id,
                'reason': 'Insufficient balance'
            })

# 🔍 Query service for reporting
class QueryService:
    def __init__(self, services: List[ServiceBase]):
        self.services = services
        print("🔍 Query service ready!")
    
    def get_customer_orders(self, customer_id: str) -> List[Dict[str, Any]]:
        orders = []
        for service in self.services:
            if isinstance(service, OrderService):
                for order_id, order_data in service.read_model.items():
                    if order_data['customer_id'] == customer_id:
                        orders.append(order_data)
        return orders
    
    def get_service_events(self, service_name: str) -> List[Dict[str, Any]]:
        for service in self.services:
            if service.name == service_name:
                return service.event_store
        return []

# 🛡️ Specification for business rules
class CustomerCreditSpec(Specification):
    def __init__(self, min_balance: float):
        self.min_balance = min_balance
    
    def is_satisfied_by(self, customer: Dict[str, Any]) -> bool:
        return customer.get('balance', 0) >= self.min_balance

# 🎮 Run the microservices system!
message_bus = MessageBus()
order_service = OrderService(message_bus)
payment_service = PaymentService(message_bus)
query_service = QueryService([order_service, payment_service])

# Place some orders
print("\n🚀 Starting microservices demo...\n")

# Customer with sufficient balance
order1 = order_service.place_order('cust-123', [
    {'name': 'Microservices Book', 'price': 49.99, 'quantity': 1, 'emoji': '📘'},
    {'name': 'Coffee', 'price': 4.99, 'quantity': 2, 'emoji': '☕'}
])

time.sleep(0.1)  # Simulate async processing

# Customer with insufficient balance
order2 = order_service.place_order('cust-456', [
    {'name': 'Cloud Architecture Course', 'price': 299.99, 'quantity': 1, 'emoji': '☁️'}
])

# Query customer orders
print(f"\n📊 Customer cust-123 orders:")
for order in query_service.get_customer_orders('cust-123'):
    print(f"  📦 Order {order['id']}: ${order['total']} ({order['item_count']} items)")

# Show event sourcing
print(f"\n🎬 Payment service events:")
for event in query_service.get_service_events("PaymentService"):
    print(f"  📝 {event['type']} at {event['timestamp'].strftime('%H:%M:%S')}")

🎓 Key Takeaways

You’ve mastered advanced design patterns! Here’s what you can now do:

• ✅ Implement Repository pattern for clean data access 💪
• ✅ Apply CQRS to separate read and write concerns 🛡️
• ✅ Use Event Sourcing for complete audit trails 🎯
• ✅ Build Sagas for distributed transactions 🐛
• ✅ Create flexible systems with advanced patterns! 🚀

Remember: Advanced patterns are powerful tools, but use them wisely! Start simple and evolve your architecture as needed. 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered advanced design patterns!

Here’s what to do next:

1. 💻 Practice implementing these patterns in your projects
2. 🏗️ Build a small microservices system using CQRS and Event Sourcing
3. 📚 Move on to our next tutorial: Parallel Programming Foundations
4. 🌟 Share your architectural insights with the community!

Remember: Every architect started with simple patterns. Keep building, keep learning, and most importantly, design systems that solve real problems! 🚀

Happy architecting! 🎉🚀✨