// 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.
Comprehensive prompting guide for LTX-2 video generation model. Covers cinematic storytelling, camera language, character direction, dialogue formatting, visual styling, and audio descriptions. Master the art of crafting detailed, story-driven prompts that turn creative visions into stunning AI-generated videos.
Design and run A/B tests that produce valid, actionable results. Covers hypothesis design, sample size, metrics selection, and analysis.
Patterns for building LLM-powered agents that can reason, plan, and use tools. Covers agent architectures (ReAct, Plan-and-Execute), tool design, multi-agent systems, memory management, and production deployment considerations.
Decision frameworks for using AI Skills effectively. When to invoke skills, how to compose them, and integration patterns.
Set up analytics tracking that informs decisions. Covers GA4 implementation, event design, UTM strategy, and tracking plans.
REST API design principles and best practices. Use when designing new APIs, reviewing API contracts, or improving existing endpoints. Covers resource modeling, HTTP semantics, error handling, versioning, and documentation.
| 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"}} |
���������������������������������������������������������������������
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" %}
���������������������������������������������
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.
{% 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
# 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" %}
������������������
����������¶ HTTP Adapter
��������,���������
��������¼���������
Driving
Adapters PORTS À���� Interfaces
(Application)
��������,���������
��������¼���������
����������¶ CLI Adapter
������������������
������������������
DB Adapter À��
������������������
Driven
������������������ Adapters
Email Adapter À��
������������������
{% 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" %}
| 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 |
{% 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 %}
���������������������������������������������������������������������
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" %}
Decompose by:
1. Business Capability Orders, Inventory, Shipping
2. Subdomain (DDD) Core, Supporting, Generic
3. Team Structure Conway's Law
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
{% 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" %}
�����������������������������������������
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 %}
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 %}
| 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 |