Core Architecture Patterns: A Comprehensive Guide with Python Examples

Personal Knowledge Sharing: My journey through architectural patterns that have shaped modern software development

Architecture patterns are the backbone of scalable, maintainable software systems. Through my years of building production systems, I've encountered various architectural challenges that led me to explore and implement different patterns. This guide shares my personal experience with four fundamental architecture patterns, complete with Python implementations and real-world insights.

Table of Contents

1. Clean Architecture / Hexagonal Architecture

2. Layered Architecture

3. CQRS (Command Query Responsibility Segregation)

4. Saga Pattern

---

1. Clean Architecture / Hexagonal Architecture

Overview

Clean Architecture, also known as Hexagonal Architecture or Ports and Adapters pattern, was a game-changer in my development journey. It solved the problem of tightly coupled code that was difficult to test and maintain.

Key Principles

1. Dependency Inversion: Dependencies point inward toward the business logic

2. Independence: Business rules don't depend on external frameworks

3. Testability: Business logic can be tested in isolation

4. Flexibility: Easy to change external dependencies

Python Implementation

# Domain Layer - Core Business Logic
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class User:
    id: Optional[int]
    email: str
    name: str
    created_at: Optional[datetime] = None

class UserRepository(ABC):
    """Port - Interface for user data access"""
    
    @abstractmethod
    def save(self, user: User) -> User:
        pass
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        pass
    
    @abstractmethod
    def find_by_email(self, email: str) -> Optional[User]:
        pass

class EmailService(ABC):
    """Port - Interface for email notifications"""
    
    @abstractmethod
    def send_welcome_email(self, user: User) -> bool:
        pass

# Application Layer - Use Cases
class UserRegistrationUseCase:
    def __init__(self, user_repo: UserRepository, email_service: EmailService):
        self._user_repo = user_repo
        self._email_service = email_service
    
    def register_user(self, email: str, name: str) -> User:
        # Business logic
        existing_user = self._user_repo.find_by_email(email)
        if existing_user:
            raise ValueError("User already exists")
        
        user = User(id=None, email=email, name=name)
        saved_user = self._user_repo.save(user)
        
        # Send welcome email
        self._email_service.send_welcome_email(saved_user)
        
        return saved_user

# Infrastructure Layer - Adapters
import sqlite3
from typing import Optional

class SQLiteUserRepository(UserRepository):
    """Adapter - Concrete implementation for SQLite"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    email TEXT UNIQUE NOT NULL,
                    name TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
    
    def save(self, user: User) -> User:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "INSERT INTO users (email, name) VALUES (?, ?)",
                (user.email, user.name)
            )
            user.id = cursor.lastrowid
            user.created_at = datetime.now()
        return user
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT id, email, name, created_at FROM users WHERE id = ?",
                (user_id,)
            ).fetchone()
            
            if row:
                return User(
                    id=row[0], 
                    email=row[1], 
                    name=row[2], 
                    created_at=datetime.fromisoformat(row[3])
                )
        return None
    
    def find_by_email(self, email: str) -> Optional[User]:
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT id, email, name, created_at FROM users WHERE email = ?",
                (email,)
            ).fetchone()
            
            if row:
                return User(
                    id=row[0], 
                    email=row[1], 
                    name=row[2], 
                    created_at=datetime.fromisoformat(row[3])
                )
        return None

class SMTPEmailService(EmailService):
    """Adapter - Concrete implementation for SMTP"""
    
    def __init__(self, smtp_server: str, port: int, username: str, password: str):
        self.smtp_server = smtp_server
        self.port = port
        self.username = username
        self.password = password
    
    def send_welcome_email(self, user: User) -> bool:
        # Simplified implementation
        print(f"Sending welcome email to {user.email}")
        # In real implementation, use smtplib
        return True

# Interface Layer - Controllers
from flask import Flask, request, jsonify

class UserController:
    def __init__(self, user_registration_use_case: UserRegistrationUseCase):
        self.user_registration_use_case = user_registration_use_case
    
    def register(self):
        try:
            data = request.get_json()
            user = self.user_registration_use_case.register_user(
                email=data['email'],
                name=data['name']
            )
            return jsonify({
                'id': user.id,
                'email': user.email,
                'name': user.name,
                'created_at': user.created_at.isoformat()
            }), 201
        except ValueError as e:
            return jsonify({'error': str(e)}), 400
        except Exception as e:
            return jsonify({'error': 'Internal server error'}), 500

# Dependency Injection Setup
def create_app():
    app = Flask(__name__)
    
    # Infrastructure setup
    user_repo = SQLiteUserRepository("users.db")
    email_service = SMTPEmailService("smtp.gmail.com", 587, "user", "pass")
    
    # Application setup
    user_registration_use_case = UserRegistrationUseCase(user_repo, email_service)
    
    # Interface setup
    user_controller = UserController(user_registration_use_case)
    
    app.add_url_rule('/users', 'register_user', user_controller.register, methods=['POST'])
    
    return app

Sequence Diagram

Testing Strategy

import unittest
from unittest.mock import Mock

class TestUserRegistrationUseCase(unittest.TestCase):
    def setUp(self):
        self.mock_user_repo = Mock(spec=UserRepository)
        self.mock_email_service = Mock(spec=EmailService)
        self.use_case = UserRegistrationUseCase(
            self.mock_user_repo, 
            self.mock_email_service
        )
    
    def test_register_new_user_success(self):
        # Arrange
        self.mock_user_repo.find_by_email.return_value = None
        self.mock_user_repo.save.return_value = User(
            id=1, email="[email protected]", name="Test User"
        )
        self.mock_email_service.send_welcome_email.return_value = True
        
        # Act
        result = self.use_case.register_user("[email protected]", "Test User")
        
        # Assert
        self.assertEqual(result.email, "[email protected]")
        self.mock_user_repo.save.assert_called_once()
        self.mock_email_service.send_welcome_email.assert_called_once()
    
    def test_register_existing_user_raises_error(self):
        # Arrange
        existing_user = User(id=1, email="[email protected]", name="Existing User")
        self.mock_user_repo.find_by_email.return_value = existing_user
        
        # Act & Assert
        with self.assertRaises(ValueError) as context:
            self.use_case.register_user("[email protected]", "Test User")
        
        self.assertEqual(str(context.exception), "User already exists")

---

2. Layered Architecture

Overview

Layered Architecture (N-tier) was one of the first patterns I learned. It's straightforward but taught me the importance of separation of concerns.

Traditional Layers

1. Presentation Layer: User interface and input handling

2. Business/Service Layer: Business logic and rules

3. Data Access Layer: Data persistence and retrieval

4. Database Layer: Data storage

Python Implementation

# Data Layer
import sqlite3
from typing import List, Optional
from dataclasses import dataclass

@dataclass
class Product:
    id: Optional[int]
    name: str
    price: float
    category: str
    stock: int

class ProductDataAccess:
    """Data Access Layer - Handles database operations"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    price REAL NOT NULL,
                    category TEXT NOT NULL,
                    stock INTEGER NOT NULL
                )
            """)
    
    def get_all_products(self) -> List[Product]:
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT id, name, price, category, stock FROM products"
            ).fetchall()
            
            return [Product(id=row[0], name=row[1], price=row[2], 
                          category=row[3], stock=row[4]) for row in rows]
    
    def get_product_by_id(self, product_id: int) -> Optional[Product]:
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT id, name, price, category, stock FROM products WHERE id = ?",
                (product_id,)
            ).fetchone()
            
            if row:
                return Product(id=row[0], name=row[1], price=row[2], 
                             category=row[3], stock=row[4])
        return None
    
    def create_product(self, product: Product) -> Product:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "INSERT INTO products (name, price, category, stock) VALUES (?, ?, ?, ?)",
                (product.name, product.price, product.category, product.stock)
            )
            product.id = cursor.lastrowid
        return product
    
    def update_stock(self, product_id: int, new_stock: int) -> bool:
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "UPDATE products SET stock = ? WHERE id = ?",
                (new_stock, product_id)
            )
            return cursor.rowcount > 0

# Business Layer
class ProductService:
    """Business Logic Layer - Contains business rules"""
    
    def __init__(self, data_access: ProductDataAccess):
        self.data_access = data_access
    
    def get_available_products(self) -> List[Product]:
        """Get products that are in stock"""
        all_products = self.data_access.get_all_products()
        return [p for p in all_products if p.stock > 0]
    
    def get_products_by_category(self, category: str) -> List[Product]:
        """Get products filtered by category"""
        all_products = self.data_access.get_all_products()
        return [p for p in all_products if p.category.lower() == category.lower()]
    
    def create_product(self, name: str, price: float, category: str, stock: int) -> Product:
        """Create a new product with business validation"""
        # Business rules
        if price <= 0:
            raise ValueError("Price must be positive")
        if stock < 0:
            raise ValueError("Stock cannot be negative")
        if len(name.strip()) == 0:
            raise ValueError("Product name cannot be empty")
        
        product = Product(id=None, name=name.strip(), price=price, 
                         category=category.strip(), stock=stock)
        return self.data_access.create_product(product)
    
    def purchase_product(self, product_id: int, quantity: int) -> bool:
        """Handle product purchase with business logic"""
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        
        product = self.data_access.get_product_by_id(product_id)
        if not product:
            raise ValueError("Product not found")
        
        if product.stock < quantity:
            raise ValueError("Insufficient stock")
        
        new_stock = product.stock - quantity
        return self.data_access.update_stock(product_id, new_stock)
    
    def calculate_order_total(self, items: List[tuple]) -> float:
        """Calculate total price for multiple items"""
        total = 0.0
        for product_id, quantity in items:
            product = self.data_access.get_product_by_id(product_id)
            if product:
                total += product.price * quantity
        return total

# Presentation Layer
from flask import Flask, request, jsonify

class ProductController:
    """Presentation Layer - Handles HTTP requests"""
    
    def __init__(self, product_service: ProductService):
        self.product_service = product_service
    
    def get_products(self):
        """GET /products - Get all available products"""
        try:
            category = request.args.get('category')
            if category:
                products = self.product_service.get_products_by_category(category)
            else:
                products = self.product_service.get_available_products()
            
            return jsonify([{
                'id': p.id,
                'name': p.name,
                'price': p.price,
                'category': p.category,
                'stock': p.stock
            } for p in products])
        except Exception as e:
            return jsonify({'error': str(e)}), 500
    
    def create_product(self):
        """POST /products - Create new product"""
        try:
            data = request.get_json()
            product = self.product_service.create_product(
                name=data['name'],
                price=data['price'],
                category=data['category'],
                stock=data['stock']
            )
            
            return jsonify({
                'id': product.id,
                'name': product.name,
                'price': product.price,
                'category': product.category,
                'stock': product.stock
            }), 201
        except ValueError as e:
            return jsonify({'error': str(e)}), 400
        except Exception as e:
            return jsonify({'error': 'Internal server error'}), 500
    
    def purchase_product(self):
        """POST /products/{id}/purchase - Purchase product"""
        try:
            product_id = int(request.view_args['id'])
            data = request.get_json()
            quantity = data['quantity']
            
            success = self.product_service.purchase_product(product_id, quantity)
            
            if success:
                return jsonify({'message': 'Purchase successful'}), 200
            else:
                return jsonify({'error': 'Purchase failed'}), 400
        except ValueError as e:
            return jsonify({'error': str(e)}), 400
        except Exception as e:
            return jsonify({'error': 'Internal server error'}), 500

# Application Setup
def create_layered_app():
    app = Flask(__name__)
    
    # Layer initialization
    data_access = ProductDataAccess("products.db")
    product_service = ProductService(data_access)
    product_controller = ProductController(product_service)
    
    # Route registration
    app.add_url_rule('/products', 'get_products', 
                     product_controller.get_products, methods=['GET'])
    app.add_url_rule('/products', 'create_product', 
                     product_controller.create_product, methods=['POST'])
    app.add_url_rule('/products/<int:id>/purchase', 'purchase_product', 
                     product_controller.purchase_product, methods=['POST'])
    
    return app

Sequence Diagram

When to Use vs Other Patterns

Use Layered Architecture when:

• Building traditional CRUD applications

• Team is familiar with the pattern

• Requirements are stable and well-defined

• Simplicity is preferred over flexibility

Consider other patterns when:

• Need high testability (Clean Architecture)

• Complex business logic (Domain-Driven Design)

• Different read/write requirements (CQRS)

---

3. CQRS (Command Query Responsibility Segregation)

Overview

CQRS revolutionized how I think about data operations. It separates read and write operations, allowing optimization for each use case.

Key Concepts

• Commands: Change state, don't return data

• Queries: Return data, don't change state

• Separate Models: Different models for reads and writes

• Event Sourcing: Optional, but common integration

Python Implementation

# Domain Events
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
import uuid

@dataclass
class DomainEvent:
    event_id: str
    aggregate_id: str
    event_type: str
    timestamp: datetime
    data: dict

# Command Side - Write Model
@dataclass
class CreateOrderCommand:
    customer_id: str
    items: List[dict]
    total_amount: float

@dataclass
class AddOrderItemCommand:
    order_id: str
    product_id: str
    quantity: int
    price: float

@dataclass
class Order:
    id: str
    customer_id: str
    items: List[dict]
    total_amount: float
    status: str
    created_at: datetime
    
    def add_item(self, product_id: str, quantity: int, price: float):
        """Business logic for adding items"""
        item = {
            'product_id': product_id,
            'quantity': quantity,
            'price': price,
            'total': quantity * price
        }
        self.items.append(item)
        self.total_amount += item['total']
        
        # Generate domain event
        return DomainEvent(
            event_id=str(uuid.uuid4()),
            aggregate_id=self.id,
            event_type='OrderItemAdded',
            timestamp=datetime.now(),
            data=item
        )

class OrderCommandRepository:
    """Write-side repository"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS orders (
                    id TEXT PRIMARY KEY,
                    customer_id TEXT NOT NULL,
                    total_amount REAL NOT NULL,
                    status TEXT NOT NULL,
                    created_at TIMESTAMP NOT NULL
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS order_items (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    order_id TEXT NOT NULL,
                    product_id TEXT NOT NULL,
                    quantity INTEGER NOT NULL,
                    price REAL NOT NULL,
                    FOREIGN KEY (order_id) REFERENCES orders (id)
                )
            """)
    
    def save(self, order: Order) -> None:
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            # Save order
            conn.execute("""
                INSERT OR REPLACE INTO orders 
                (id, customer_id, total_amount, status, created_at) 
                VALUES (?, ?, ?, ?, ?)
            """, (order.id, order.customer_id, order.total_amount, 
                  order.status, order.created_at))
            
            # Save items
            conn.execute("DELETE FROM order_items WHERE order_id = ?", (order.id,))
            for item in order.items:
                conn.execute("""
                    INSERT INTO order_items 
                    (order_id, product_id, quantity, price) 
                    VALUES (?, ?, ?, ?)
                """, (order.id, item['product_id'], 
                      item['quantity'], item['price']))
    
    def get_by_id(self, order_id: str) -> Optional[Order]:
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            # Get order
            order_row = conn.execute("""
                SELECT id, customer_id, total_amount, status, created_at 
                FROM orders WHERE id = ?
            """, (order_id,)).fetchone()
            
            if not order_row:
                return None
            
            # Get items
            item_rows = conn.execute("""
                SELECT product_id, quantity, price 
                FROM order_items WHERE order_id = ?
            """, (order_id,)).fetchall()
            
            items = [{
                'product_id': row[0],
                'quantity': row[1],
                'price': row[2],
                'total': row[1] * row[2]
            } for row in item_rows]
            
            return Order(
                id=order_row[0],
                customer_id=order_row[1],
                total_amount=order_row[2],
                status=order_row[3],
                created_at=datetime.fromisoformat(order_row[4]),
                items=items
            )

# Command Handlers
class OrderCommandHandler:
    def __init__(self, repository: OrderCommandRepository, event_bus):
        self.repository = repository
        self.event_bus = event_bus
    
    def handle_create_order(self, command: CreateOrderCommand) -> str:
        order_id = str(uuid.uuid4())
        order = Order(
            id=order_id,
            customer_id=command.customer_id,
            items=command.items,
            total_amount=command.total_amount,
            status='pending',
            created_at=datetime.now()
        )
        
        self.repository.save(order)
        
        # Publish event
        event = DomainEvent(
            event_id=str(uuid.uuid4()),
            aggregate_id=order_id,
            event_type='OrderCreated',
            timestamp=datetime.now(),
            data={
                'customer_id': command.customer_id,
                'total_amount': command.total_amount
            }
        )
        self.event_bus.publish(event)
        
        return order_id
    
    def handle_add_order_item(self, command: AddOrderItemCommand) -> None:
        order = self.repository.get_by_id(command.order_id)
        if not order:
            raise ValueError("Order not found")
        
        event = order.add_item(command.product_id, command.quantity, command.price)
        self.repository.save(order)
        self.event_bus.publish(event)

# Query Side - Read Model
@dataclass
class OrderSummary:
    id: str
    customer_id: str
    customer_name: str
    total_amount: float
    item_count: int
    status: str
    created_at: datetime

@dataclass
class OrderDetails:
    id: str
    customer_id: str
    customer_name: str
    items: List[dict]
    total_amount: float
    status: str
    created_at: datetime

class OrderQueryRepository:
    """Read-side repository with optimized views"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_read_db()
    
    def _init_read_db(self):
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS order_summaries (
                    id TEXT PRIMARY KEY,
                    customer_id TEXT NOT NULL,
                    customer_name TEXT NOT NULL,
                    total_amount REAL NOT NULL,
                    item_count INTEGER NOT NULL,
                    status TEXT NOT NULL,
                    created_at TIMESTAMP NOT NULL
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS order_details_view (
                    order_id TEXT NOT NULL,
                    customer_id TEXT NOT NULL,
                    customer_name TEXT NOT NULL,
                    product_id TEXT NOT NULL,
                    product_name TEXT NOT NULL,
                    quantity INTEGER NOT NULL,
                    price REAL NOT NULL,
                    total_amount REAL NOT NULL,
                    status TEXT NOT NULL,
                    created_at TIMESTAMP NOT NULL
                )
            """)
    
    def get_order_summaries(self, customer_id: Optional[str] = None) -> List[OrderSummary]:
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            if customer_id:
                rows = conn.execute("""
                    SELECT id, customer_id, customer_name, total_amount, 
                           item_count, status, created_at 
                    FROM order_summaries WHERE customer_id = ?
                    ORDER BY created_at DESC
                """, (customer_id,)).fetchall()
            else:
                rows = conn.execute("""
                    SELECT id, customer_id, customer_name, total_amount, 
                           item_count, status, created_at 
                    FROM order_summaries 
                    ORDER BY created_at DESC
                """).fetchall()
            
            return [OrderSummary(
                id=row[0],
                customer_id=row[1],
                customer_name=row[2],
                total_amount=row[3],
                item_count=row[4],
                status=row[5],
                created_at=datetime.fromisoformat(row[6])
            ) for row in rows]
    
    def get_order_details(self, order_id: str) -> Optional[OrderDetails]:
        import sqlite3
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute("""
                SELECT order_id, customer_id, customer_name, product_id, 
                       product_name, quantity, price, total_amount, status, created_at
                FROM order_details_view WHERE order_id = ?
            """, (order_id,)).fetchall()
            
            if not rows:
                return None
            
            first_row = rows[0]
            items = [{
                'product_id': row[3],
                'product_name': row[4],
                'quantity': row[5],
                'price': row[6]
            } for row in rows]
            
            return OrderDetails(
                id=first_row[0],
                customer_id=first_row[1],
                customer_name=first_row[2],
                items=items,
                total_amount=first_row[7],
                status=first_row[8],
                created_at=datetime.fromisoformat(first_row[9])
            )

# Query Handlers
class OrderQueryHandler:
    def __init__(self, repository: OrderQueryRepository):
        self.repository = repository
    
    def get_customer_orders(self, customer_id: str) -> List[OrderSummary]:
        return self.repository.get_order_summaries(customer_id)
    
    def get_all_orders(self) -> List[OrderSummary]:
        return self.repository.get_order_summaries()
    
    def get_order_details(self, order_id: str) -> Optional[OrderDetails]:
        return self.repository.get_order_details(order_id)

# Event Bus for Read Model Updates
class SimpleEventBus:
    def __init__(self):
        self.handlers = {}
    
    def subscribe(self, event_type: str, handler):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    def publish(self, event: DomainEvent):
        handlers = self.handlers.get(event.event_type, [])
        for handler in handlers:
            handler(event)

# Read Model Projection
class OrderProjection:
    def __init__(self, query_repo: OrderQueryRepository):
        self.query_repo = query_repo
    
    def handle_order_created(self, event: DomainEvent):
        """Update read model when order is created"""
        import sqlite3
        with sqlite3.connect(self.query_repo.db_path) as conn:
            conn.execute("""
                INSERT INTO order_summaries 
                (id, customer_id, customer_name, total_amount, item_count, status, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (
                event.aggregate_id,
                event.data['customer_id'],
                f"Customer {event.data['customer_id']}",  # Simplified
                event.data['total_amount'],
                0,  # Will be updated when items are added
                'pending',
                event.timestamp
            ))

# Controllers
from flask import Flask, request, jsonify

class OrderCommandController:
    def __init__(self, command_handler: OrderCommandHandler):
        self.command_handler = command_handler
    
    def create_order(self):
        try:
            data = request.get_json()
            command = CreateOrderCommand(
                customer_id=data['customer_id'],
                items=data['items'],
                total_amount=data['total_amount']
            )
            order_id = self.command_handler.handle_create_order(command)
            return jsonify({'order_id': order_id}), 201
        except Exception as e:
            return jsonify({'error': str(e)}), 400

class OrderQueryController:
    def __init__(self, query_handler: OrderQueryHandler):
        self.query_handler = query_handler
    
    def get_orders(self):
        try:
            customer_id = request.args.get('customer_id')
            if customer_id:
                orders = self.query_handler.get_customer_orders(customer_id)
            else:
                orders = self.query_handler.get_all_orders()
            
            return jsonify([{
                'id': o.id,
                'customer_id': o.customer_id,
                'customer_name': o.customer_name,
                'total_amount': o.total_amount,
                'item_count': o.item_count,
                'status': o.status,
                'created_at': o.created_at.isoformat()
            } for o in orders])
        except Exception as e:
            return jsonify({'error': str(e)}), 500
    
    def get_order_details(self, order_id: str):
        try:
            order = self.query_handler.get_order_details(order_id)
            if not order:
                return jsonify({'error': 'Order not found'}), 404
            
            return jsonify({
                'id': order.id,
                'customer_id': order.customer_id,
                'customer_name': order.customer_name,
                'items': order.items,
                'total_amount': order.total_amount,
                'status': order.status,
                'created_at': order.created_at.isoformat()
            })
        except Exception as e:
            return jsonify({'error': str(e)}), 500

Sequence Diagram

---

4. Saga Pattern

Overview

The Saga pattern solved distributed transaction challenges in my microservices architecture. It manages long-running business processes across multiple services.

Key Concepts

• Choreography: Services communicate through events

• Orchestration: Central coordinator manages the saga

• Compensation: Rollback actions for failed steps

Implementation with AWS Step Functions

# Saga Orchestrator using AWS Step Functions
import json
import boto3
from dataclasses import dataclass, asdict
from typing import Dict, Any, Optional
from enum import Enum

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

@dataclass
class SagaState:
    saga_id: str
    status: SagaStatus
    current_step: int
    steps_completed: list
    compensation_data: Dict[str, Any]
    error_message: Optional[str] = None

@dataclass
class OrderSaga:
    order_id: str
    customer_id: str
    items: list
    total_amount: float
    payment_id: Optional[str] = None
    inventory_reservation_id: Optional[str] = None
    shipping_id: Optional[str] = None

# Step Functions State Machine Definition
SAGA_STATE_MACHINE_DEFINITION = {
    "Comment": "Order Processing Saga",
    "StartAt": "ReserveInventory",
    "States": {
        "ReserveInventory": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:region:account:function:reserve-inventory",
            "Retry": [
                {
                    "ErrorEquals": ["States.TaskFailed"],
                    "IntervalSeconds": 2,
                    "MaxAttempts": 3,
                    "BackoffRate": 2.0
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": ["States.ALL"],
                    "Next": "SagaFailed"
                }
            ],
            "Next": "ProcessPayment"
        },
        "ProcessPayment": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:region:account:function:process-payment",
            "Retry": [
                {
                    "ErrorEquals": ["States.TaskFailed"],
                    "IntervalSeconds": 2,
                    "MaxAttempts": 3,
                    "BackoffRate": 2.0
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": ["States.ALL"],
                    "Next": "CompensateInventory"
                }
            ],
            "Next": "CreateShipment"
        },
        "CreateShipment": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:region:account:function:create-shipment",
            "Retry": [
                {
                    "ErrorEquals": ["States.TaskFailed"],
                    "IntervalSeconds": 2,
                    "MaxAttempts": 3,
                    "BackoffRate": 2.0
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": ["States.ALL"],
                    "Next": "CompensatePayment"
                }
            ],
            "Next": "SagaCompleted"
        },
        "SagaCompleted": {
            "Type": "Pass",
            "Result": {"status": "completed"},
            "End": True
        },
        "CompensateInventory": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:region:account:function:compensate-inventory",
            "Next": "SagaFailed"
        },
        "CompensatePayment": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:region:account:function:compensate-payment",
            "Next": "CompensateInventory"
        },
        "SagaFailed": {
            "Type": "Pass",
            "Result": {"status": "failed"},
            "End": True
        }
    }
}

# Saga Orchestrator
class OrderSagaOrchestrator:
    def __init__(self, region_name: str = 'us-east-1'):
        self.stepfunctions = boto3.client('stepfunctions', region_name=region_name)
        self.state_machine_arn = None  # Set this to your actual ARN
    
    def start_saga(self, order_saga: OrderSaga) -> str:
        """Start the order processing saga"""
        execution_name = f"order-saga-{order_saga.order_id}"
        
        response = self.stepfunctions.start_execution(
            stateMachineArn=self.state_machine_arn,
            name=execution_name,
            input=json.dumps(asdict(order_saga))
        )
        
        return response['executionArn']
    
    def get_saga_status(self, execution_arn: str) -> Dict[str, Any]:
        """Get current status of the saga"""
        response = self.stepfunctions.describe_execution(
            executionArn=execution_arn
        )
        
        return {
            'status': response['status'],
            'start_date': response['startDate'].isoformat(),
            'input': json.loads(response['input']),
            'output': json.loads(response.get('output', '{}')),
            'error': response.get('error', '')
        }

# Lambda Functions for Saga Steps
def reserve_inventory_lambda(event, context):
    """Lambda function to reserve inventory"""
    try:
        order_data = event
        
        # Simulate inventory service call
        inventory_service = InventoryService()
        reservation_id = inventory_service.reserve_items(
            order_data['items'], 
            order_data['order_id']
        )
        
        # Update saga state
        order_data['inventory_reservation_id'] = reservation_id
        
        return {
            'statusCode': 200,
            'body': order_data
        }
    except Exception as e:
        raise Exception(f"Inventory reservation failed: {str(e)}")

def process_payment_lambda(event, context):
    """Lambda function to process payment"""
    try:
        order_data = event['body'] if 'body' in event else event
        
        # Simulate payment service call
        payment_service = PaymentService()
        payment_id = payment_service.charge_customer(
            order_data['customer_id'],
            order_data['total_amount'],
            order_data['order_id']
        )
        
        # Update saga state
        order_data['payment_id'] = payment_id
        
        return {
            'statusCode': 200,
            'body': order_data
        }
    except Exception as e:
        raise Exception(f"Payment processing failed: {str(e)}")

def create_shipment_lambda(event, context):
    """Lambda function to create shipment"""
    try:
        order_data = event['body'] if 'body' in event else event
        
        # Simulate shipping service call
        shipping_service = ShippingService()
        shipping_id = shipping_service.create_shipment(
            order_data['customer_id'],
            order_data['items'],
            order_data['order_id']
        )
        
        # Update saga state
        order_data['shipping_id'] = shipping_id
        
        return {
            'statusCode': 200,
            'body': order_data
        }
    except Exception as e:
        raise Exception(f"Shipment creation failed: {str(e)}")

# Compensation Functions
def compensate_inventory_lambda(event, context):
    """Lambda function to compensate inventory reservation"""
    try:
        order_data = event['body'] if 'body' in event else event
        
        if 'inventory_reservation_id' in order_data:
            inventory_service = InventoryService()
            inventory_service.release_reservation(
                order_data['inventory_reservation_id']
            )
        
        return {
            'statusCode': 200,
            'body': {'compensation': 'inventory_released'}
        }
    except Exception as e:
        print(f"Inventory compensation failed: {str(e)}")
        return {
            'statusCode': 500,
            'body': {'error': str(e)}
        }

def compensate_payment_lambda(event, context):
    """Lambda function to compensate payment"""
    try:
        order_data = event['body'] if 'body' in event else event
        
        if 'payment_id' in order_data:
            payment_service = PaymentService()
            payment_service.refund_payment(
                order_data['payment_id']
            )
        
        return {
            'statusCode': 200,
            'body': {'compensation': 'payment_refunded'}
        }
    except Exception as e:
        print(f"Payment compensation failed: {str(e)}")
        return {
            'statusCode': 500,
            'body': {'error': str(e)}
        }

# Service Implementations (Simplified)
class InventoryService:
    def reserve_items(self, items: list, order_id: str) -> str:
        """Reserve inventory items"""
        # Simulate business logic
        for item in items:
            if not self._check_availability(item['product_id'], item['quantity']):
                raise Exception(f"Insufficient inventory for product {item['product_id']}")
        
        reservation_id = f"res-{order_id}-{hash(str(items))}"
        # Store reservation in database
        return reservation_id
    
    def _check_availability(self, product_id: str, quantity: int) -> bool:
        # Simulate inventory check
        return True  # Simplified
    
    def release_reservation(self, reservation_id: str) -> None:
        """Release inventory reservation"""
        # Remove reservation from database
        print(f"Released inventory reservation: {reservation_id}")

class PaymentService:
    def charge_customer(self, customer_id: str, amount: float, order_id: str) -> str:
        """Charge customer payment"""
        # Simulate payment processing
        if amount <= 0:
            raise Exception("Invalid payment amount")
        
        payment_id = f"pay-{customer_id}-{order_id}"
        # Store payment record
        return payment_id
    
    def refund_payment(self, payment_id: str) -> None:
        """Refund payment"""
        # Process refund
        print(f"Refunded payment: {payment_id}")

class ShippingService:
    def create_shipment(self, customer_id: str, items: list, order_id: str) -> str:
        """Create shipment"""
        # Simulate shipment creation
        shipping_id = f"ship-{customer_id}-{order_id}"
        # Store shipment record
        return shipping_id

# Flask Controller
from flask import Flask, request, jsonify

class SagaController:
    def __init__(self, orchestrator: OrderSagaOrchestrator):
        self.orchestrator = orchestrator
    
    def start_order_saga(self):
        """Start order processing saga"""
        try:
            data = request.get_json()
            
            order_saga = OrderSaga(
                order_id=data['order_id'],
                customer_id=data['customer_id'],
                items=data['items'],
                total_amount=data['total_amount']
            )
            
            execution_arn = self.orchestrator.start_saga(order_saga)
            
            return jsonify({
                'saga_id': execution_arn,
                'status': 'started',
                'order_id': order_saga.order_id
            }), 202
        except Exception as e:
            return jsonify({'error': str(e)}), 400
    
    def get_saga_status(self, saga_id: str):
        """Get saga execution status"""
        try:
            status = self.orchestrator.get_saga_status(saga_id)
            return jsonify(status)
        except Exception as e:
            return jsonify({'error': str(e)}), 500

# Application Setup
def create_saga_app():
    app = Flask(__name__)
    
    orchestrator = OrderSagaOrchestrator()
    saga_controller = SagaController(orchestrator)
    
    app.add_url_rule('/sagas/orders', 'start_order_saga', 
                     saga_controller.start_order_saga, methods=['POST'])
    app.add_url_rule('/sagas/<saga_id>/status', 'get_saga_status', 
                     saga_controller.get_saga_status, methods=['GET'])
    
    return app

Sequence Diagrams

Choreography vs Orchestration

Orchestration Pattern:

Choreography Pattern:

Compensation Flow:

---

Conclusion

These four architecture patterns have been instrumental in my journey as a software architect. Each serves different purposes:

• Clean Architecture: For highly testable, maintainable applications

• Layered Architecture: For traditional enterprise applications with clear separation

• CQRS: For systems with different read/write performance requirements

• Saga Pattern: For distributed systems requiring transaction management

The key is understanding when to apply each pattern. In my experience, starting with simpler patterns and evolving to more complex ones as requirements grow has been the most successful approach.

Remember, architecture patterns are tools in your toolkit. The art lies in knowing which tool to use for which problem. As systems grow and requirements change, don't be afraid to refactor and adopt patterns that better serve your needs.

Key Takeaways

1. Start Simple: Begin with patterns your team understands

2. Evolution over Revolution: Gradually introduce complexity as needed

3. Test-Driven: Ensure your architecture supports testing

4. Monitor and Measure: Use metrics to validate architectural decisions

5. Document Decisions: Capture the reasoning behind pattern choices

These patterns continue to evolve with new technologies and practices, but their fundamental principles remain solid foundations for building robust software systems.