architecture-patterns

name	architecture-patterns
version	1.0.0
description	Software architecture patterns for building maintainable, scalable systems. Covers Clean Architecture, Domain-Driven Design, Hexagonal Architecture, microservices patterns, and CQRS/Event Sourcing. Use when designing new systems or refactoring existing codebases.
license	MIT
allowed-tools	Read Edit Bash
tags	["architecture","design-patterns","ddd","clean-architecture","microservices"]
category	development/architecture
author	AI Skills Team
trigger_phrases	["system design","clean architecture","domain driven design","hexagonal architecture","microservices pattern","cqrs","event sourcing"]
variables	{"pattern":{"type":"string","description":"Architecture pattern focus","enum":["clean-architecture","hexagonal","ddd","microservices","cqrs","all"],"default":"all"},"language":{"type":"string","description":"Implementation language","enum":["python","typescript","go","java"],"default":"python"},"scale":{"type":"string","description":"System scale","enum":["small","medium","large"],"default":"medium"}}

Software Architecture Patterns

Choosing an Architecture

���������������������������������������������������������������������
                    ARCHITECTURE DECISION GUIDE                       
���������������������������������������������������������������������$
                                                                     
  Small project, simple domain?                                      
     ’ Layered Architecture (MVC)                                    
                                                                     
  Complex business logic, long-lived project?                        
     ’ Clean Architecture / Hexagonal                                
                                                                     
  Complex domain with many business rules?                           
     ’ Domain-Driven Design (DDD)                                    
                                                                     
  High traffic, independent scaling needs?                           
     ’ Microservices                                                 
                                                                     
  Different read/write patterns, event-driven?                       
     ’ CQRS / Event Sourcing                                         
                                                                     
���������������������������������������������������������������������

---

{% if pattern == "clean-architecture" or pattern == "all" %}

Clean Architecture

The Dependency Rule

    ���������������������������������������������
                  Frameworks & Drivers               UI, DB, Web
      �������������������������������������    
               Interface Adapters                 Controllers, Gateways
        �����������������������������        
               Application Layer                Use Cases
          ���������������������            
                Entities                      Business Rules
          ���������������������            
        �����������������������������        
      �������������������������������������    
    ���������������������������������������������

    Dependencies point INWARD only.
    Inner layers know nothing about outer layers.

Project Structure

{% if language == "python" %}

src/
�� domain/                    # Entities (innermost)
   �� entities/
      �� user.py
      �� order.py
   �� value_objects/
       �� email.py
�� application/               # Use Cases
   �� interfaces/           # Port definitions
      �� user_repository.py
      �� email_service.py
   �� use_cases/
       �� create_user.py
       �� process_order.py
�� infrastructure/            # Adapters (outermost)
   �� persistence/
      �� sqlalchemy_user_repo.py
   �� external/
      �� sendgrid_email.py
   �� web/
       �� fastapi_app.py
�� main.py                    # Composition root

Implementation

# domain/entities/user.py - Pure business logic, no dependencies
from dataclasses import dataclass
from domain.value_objects.email import Email

@dataclass
class User:
    id: str
    email: Email
    name: str
    is_active: bool = True

    def deactivate(self) -> None:
        if not self.is_active:
            raise ValueError("User already deactivated")
        self.is_active = False


# application/interfaces/user_repository.py - Port (abstract)
from abc import ABC, abstractmethod
from domain.entities.user import User

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> User: ...

    @abstractmethod
    def find_by_id(self, user_id: str) -> User | None: ...


# application/use_cases/create_user.py - Use Case
from dataclasses import dataclass
from application.interfaces.user_repository import UserRepository
from domain.entities.user import User
from domain.value_objects.email import Email

@dataclass
class CreateUserInput:
    email: str
    name: str

@dataclass
class CreateUserOutput:
    user_id: str

class CreateUserUseCase:
    def __init__(self, user_repo: UserRepository):
        self._user_repo = user_repo

    def execute(self, input: CreateUserInput) -> CreateUserOutput:
        email = Email(input.email)  # Validates email
        user = User(
            id=generate_id(),
            email=email,
            name=input.name
        )
        saved = self._user_repo.save(user)
        return CreateUserOutput(user_id=saved.id)


# infrastructure/persistence/sqlalchemy_user_repo.py - Adapter
from application.interfaces.user_repository import UserRepository
from domain.entities.user import User

class SqlAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self._session = session

    def save(self, user: User) -> User:
        model = UserModel.from_entity(user)
        self._session.add(model)
        self._session.flush()
        return model.to_entity()

{% elif language == "typescript" %}

// domain/entities/User.ts
export class User {
  constructor(
    public readonly id: string,
    public readonly email: Email,
    public name: string,
    public isActive: boolean = true
  ) {}

  deactivate(): void {
    if (!this.isActive) throw new Error("Already deactivated");
    this.isActive = false;
  }
}

// application/interfaces/UserRepository.ts
export interface UserRepository {
  save(user: User): Promise<User>;
  findById(id: string): Promise<User | null>;
}

// application/use-cases/CreateUser.ts
export class CreateUserUseCase {
  constructor(private userRepo: UserRepository) {}

  async execute(input: CreateUserInput): Promise<CreateUserOutput> {
    const email = new Email(input.email);
    const user = new User(generateId(), email, input.name);
    const saved = await this.userRepo.save(user);
    return { userId: saved.id };
  }
}

{% endif %} {% endif %}

{% if pattern == "hexagonal" or pattern == "all" %}

Hexagonal Architecture (Ports & Adapters)

Concept

                    ������������������
        ����������¶   HTTP Adapter   
                   ��������,���������
                            
                   ��������¼���������
   Driving                            
   Adapters               PORTS       À���� Interfaces
                     (Application)   
                                     
                   ��������,���������
                            
                   ��������¼���������
        ����������¶   CLI Adapter    
                    ������������������

                    ������������������
                       DB Adapter     À��
                    ������������������   
                                            Driven
                    ������������������    Adapters
                      Email Adapter   À��
                    ������������������

Key Concepts

• Ports: Interfaces that define how the application communicates
• Driving Adapters: Call into the application (HTTP, CLI, tests)
• Driven Adapters: Are called by the application (DB, external APIs)

{% if language == "python" %}

# ports/incoming/order_service.py (Driving Port)
from abc import ABC, abstractmethod

class OrderService(ABC):
    @abstractmethod
    def place_order(self, customer_id: str, items: list[Item]) -> Order: ...


# ports/outgoing/order_repository.py (Driven Port)
class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> Order: ...


# core/services/order_service_impl.py (Application Core)
class OrderServiceImpl(OrderService):
    def __init__(self, order_repo: OrderRepository, payment: PaymentGateway):
        self._order_repo = order_repo
        self._payment = payment

    def place_order(self, customer_id: str, items: list[Item]) -> Order:
        order = Order.create(customer_id, items)
        self._payment.charge(order.total)
        return self._order_repo.save(order)


# adapters/incoming/http/order_controller.py (Driving Adapter)
@app.post("/orders")
def create_order(request: OrderRequest, service: OrderService = Depends()):
    return service.place_order(request.customer_id, request.items)


# adapters/outgoing/persistence/postgres_order_repo.py (Driven Adapter)
class PostgresOrderRepository(OrderRepository):
    def save(self, order: Order) -> Order:
        # PostgreSQL specific implementation
        ...

{% endif %} {% endif %}

{% if pattern == "ddd" or pattern == "all" %}

Domain-Driven Design (DDD)

Building Blocks

Concept	Description	Example
Entity	Has identity, mutable	User, Order
Value Object	No identity, immutable	Email, Money
Aggregate	Consistency boundary	Order + OrderItems
Repository	Collection abstraction	UserRepository
Domain Service	Stateless operations	PricingService
Domain Event	Something that happened	OrderPlaced

Implementation

{% if language == "python" %}

# Value Object - Immutable, equality by value
@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)


# Entity - Has identity, mutable
class OrderItem:
    def __init__(self, id: str, product_id: str, quantity: int, price: Money):
        self.id = id
        self.product_id = product_id
        self.quantity = quantity
        self.price = price

    def __eq__(self, other):
        return self.id == other.id  # Identity-based equality


# Aggregate Root - Controls consistency
class Order:
    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self._items: list[OrderItem] = []
        self._status = OrderStatus.DRAFT
        self._events: list[DomainEvent] = []

    def add_item(self, product_id: str, quantity: int, price: Money) -> None:
        if self._status != OrderStatus.DRAFT:
            raise InvalidOperationError("Cannot modify confirmed order")

        item = OrderItem(
            id=generate_id(),
            product_id=product_id,
            quantity=quantity,
            price=price
        )
        self._items.append(item)

    def confirm(self) -> None:
        if not self._items:
            raise InvalidOperationError("Cannot confirm empty order")

        self._status = OrderStatus.CONFIRMED
        self._events.append(OrderConfirmed(order_id=self.id))

    @property
    def total(self) -> Money:
        return reduce(
            lambda acc, item: acc.add(item.price),
            self._items,
            Money(Decimal("0"), "USD")
        )

    def collect_events(self) -> list[DomainEvent]:
        events = self._events.copy()
        self._events.clear()
        return events


# Domain Service - Stateless operations across aggregates
class PricingService:
    def __init__(self, discount_repo: DiscountRepository):
        self._discount_repo = discount_repo

    def calculate_price(self, product: Product, customer: Customer) -> Money:
        base_price = product.price
        discount = self._discount_repo.find_for_customer(customer.id)
        if discount:
            return base_price.apply_discount(discount.percentage)
        return base_price

{% endif %}

Bounded Contexts

���������������������������������������������������������������������
                           E-COMMERCE SYSTEM                          
���������������������������������������������������������������������$
                                                                     
  �������������    �������������    �������������            
     ORDERS          INVENTORY         BILLING               
     Context          Context          Context               
                                                             
   Order            Product          Invoice                 
   OrderItem        Stock            Payment                 
   Customer*        Warehouse        Customer*               
  ������,������    ������,������    ������,������            
                                                                 
         ������������������<������������������                    
                                                                    
                   Integration Events                                
                   (OrderPlaced, StockReserved, PaymentReceived)    
                                                                     
  * Customer means different things in each context                  
    (Ubiquitous Language is context-specific)                        
                                                                     
���������������������������������������������������������������������

{% endif %}

{% if pattern == "microservices" or pattern == "all" %}

Microservices Patterns

Service Decomposition

Decompose by:
1. Business Capability  ’ Orders, Inventory, Shipping
2. Subdomain (DDD)      ’ Core, Supporting, Generic
3. Team Structure       ’ Conway's Law

Communication Patterns

Synchronous (REST/gRPC)

Order Service ��HTTP��¶ Inventory Service
              À���������
              Response

 Simple, immediate response
L Tight coupling, cascading failures

Asynchronous (Events)

Order Service ��Event��¶ Message Broker ��Event��¶ Inventory Service

 Loose coupling, resilient
L Eventual consistency, complexity

Key Patterns

{% if language == "python" %}

# API Gateway Pattern
@app.get("/orders/{order_id}")
async def get_order_details(order_id: str):
    # Aggregate from multiple services
    order = await order_service.get(order_id)
    customer = await customer_service.get(order.customer_id)
    shipment = await shipping_service.get_by_order(order_id)

    return OrderDetails(
        order=order,
        customer=customer,
        shipment=shipment
    )


# Circuit Breaker Pattern
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_inventory_service(product_id: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{INVENTORY_URL}/products/{product_id}")
        return response.json()


# Saga Pattern (Choreography)
class OrderSaga:
    async def execute(self, order: Order):
        try:
            # Step 1: Reserve inventory
            await self.events.publish(ReserveInventory(order.items))

            # Step 2: Process payment
            await self.events.publish(ProcessPayment(order.total))

            # Step 3: Confirm order
            await self.events.publish(ConfirmOrder(order.id))

        except InventoryError:
            await self.events.publish(CancelOrder(order.id))
            raise

        except PaymentError:
            await self.events.publish(ReleaseInventory(order.items))
            await self.events.publish(CancelOrder(order.id))
            raise

{% endif %} {% endif %}

{% if pattern == "cqrs" or pattern == "all" %}

CQRS & Event Sourcing

CQRS (Command Query Responsibility Segregation)

           �����������������������������������������
                         APPLICATION                
           ���������������,�������������������������
                           
         �����������������4�����������������
                                            
    ����¼����                        ����¼����
     COMMAND                           QUERY  
      MODEL                            MODEL  
    ����,����                        ����,����
                                            
    ����¼����                        ����¼����
      Write                            Read   
       DB    �������sync������������¶   DB    
    ���������                        ���������

{% if language == "python" %}

# Command Side
class CreateOrderCommand:
    customer_id: str
    items: list[OrderItemDTO]

class OrderCommandHandler:
    def __init__(self, repo: OrderRepository, events: EventPublisher):
        self._repo = repo
        self._events = events

    def handle(self, cmd: CreateOrderCommand) -> str:
        order = Order.create(cmd.customer_id, cmd.items)
        self._repo.save(order)
        self._events.publish(OrderCreated(order.id, order.total))
        return order.id


# Query Side
class OrderQueryService:
    def __init__(self, read_db: ReadDatabase):
        self._db = read_db

    def get_order_summary(self, order_id: str) -> OrderSummaryDTO:
        # Optimized read model
        return self._db.query("""
            SELECT o.id, o.status, c.name as customer_name, o.total
            FROM order_summaries o
            JOIN customers c ON o.customer_id = c.id
            WHERE o.id = ?
        """, order_id)


# Projection (keeps read model in sync)
class OrderProjection:
    @subscribe(OrderCreated)
    def on_order_created(self, event: OrderCreated):
        self._read_db.execute("""
            INSERT INTO order_summaries (id, status, total)
            VALUES (?, 'created', ?)
        """, event.order_id, event.total)

{% endif %}

Event Sourcing

Instead of storing current state, store ALL events:

Events:
1. OrderCreated { id: "123", customer: "alice", items: [...] }
2. ItemAdded { order_id: "123", product: "book", qty: 2 }
3. ItemRemoved { order_id: "123", product: "pen" }
4. OrderConfirmed { order_id: "123", total: 50.00 }

Current state = replay(events)

{% if language == "python" %}

# Event Store
class EventStore:
    def append(self, stream_id: str, events: list[Event]) -> None:
        for event in events:
            self._db.execute("""
                INSERT INTO events (stream_id, type, data, timestamp)
                VALUES (?, ?, ?, ?)
            """, stream_id, event.__class__.__name__,
                 json.dumps(event.to_dict()), datetime.utcnow())

    def get_stream(self, stream_id: str) -> list[Event]:
        rows = self._db.query(
            "SELECT type, data FROM events WHERE stream_id = ? ORDER BY id",
            stream_id
        )
        return [deserialize_event(r) for r in rows]


# Aggregate with Event Sourcing
class Order:
    def __init__(self):
        self._events: list[Event] = []
        self.id = None
        self.status = None

    @classmethod
    def from_events(cls, events: list[Event]) -> "Order":
        order = cls()
        for event in events:
            order._apply(event)
        return order

    def _apply(self, event: Event) -> None:
        if isinstance(event, OrderCreated):
            self.id = event.order_id
            self.status = "created"
        elif isinstance(event, OrderConfirmed):
            self.status = "confirmed"

    def confirm(self) -> None:
        if self.status == "confirmed":
            raise Error("Already confirmed")
        self._events.append(OrderConfirmed(self.id))
        self._apply(self._events[-1])

{% endif %} {% endif %}

---

When to Use What

Pattern	Use When	Avoid When
Clean/Hexagonal	Complex business logic, testability matters	Simple CRUD apps
DDD	Rich domain, domain experts available	Technical domains, simple logic
Microservices	Team scaling, independent deployments	Small team, startup phase
CQRS	Different read/write loads	Simple domains
Event Sourcing	Audit trail needed, temporal queries	Simple state, high write volume