# Structural Behavioral Patterns > Implements GoF structural and behavioral design patterns (Adapter, Observer, Strategy, Command, Facade, Template Method, Mediator) to decouple components and manage object responsibilities. - Skill: `paulpas/structural-behavioral-patterns` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/structural-behavioral-patterns` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/structural-behavioral-patterns/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/structural-behavioral-patterns --- # Structural & Behavioral Design Patterns Applies GoF structural and behavioral design patterns to decouple components, manage object responsibilities, and establish clean communication channels between classes. This skill makes the model select the right pattern for interface adaptation (Adapter), feature composition (Bridge, Composite, Decorator, Facade, Proxy, Flyweight), runtime behavior swapping (Strategy), request encapsulation (Command), template-based inheritance (Template Method), event-driven decoupling (Observer), and workflow orchestration (Mediator, Chain of Responsibility). ## TL;DR Checklist - [ ] Identify the structural problem first: incompatible interfaces (Adapter), tree structures (Composite), dynamic augmentation (Decorator), simplified APIs (Facade), access control (Proxy), shared data (Flyweight) - [ ] Identify the behavioral problem first: interchangeable algorithms (Strategy), request encapsulation with undo (Command), template-based inheritance (Template Method), event pub/sub (Observer), workflow delegation (Mediator), sequential processing (Chain of Responsibility) - [ ] Prefer composition over inheritance for all structural and behavioral patterns - [ ] Use `typing.Protocol` or ABC interfaces to ensure transparent substitution of pattern participants - [ ] Enforce Open/Closed Principle through polymorphism, not conditional logic branches - [ ] Ensure Facade does not become a god object — delegate to domain services - [ ] Keep Command objects immutable where possible for thread-safe undo/redo stacks - [ ] Use Observer with weak references to prevent memory leaks from forgotten unsubscription - [ ] Avoid over-engineering: do not apply patterns where simple function calls suffice --- ## When to Use Use this skill when: - Two existing classes have incompatible interfaces and you need them to work together without modifying either class's source - You need to add new responsibilities to objects dynamically at runtime instead of via subclassing (Decorator) - A subsystem has a complex API and you want a simplified unified interface for common operations (Facade) - You want to swap algorithms or strategies at runtime based on context or configuration (Strategy) - You need to encapsulate requests as objects so you can parameterize clients with different requests, queue them, log them, or support undo/redo (Command) - Multiple objects need to be notified when another object's state changes, and you want loose coupling between the subject and its observers (Observer) - You have a family of related operations that share common steps but differ in specific details (Template Method) - You are building a tree-like part-whole hierarchy where clients should treat individual objects and compositions uniformly (Composite) - You need to control access to an object, defer initialization, or add caching/authentication layers without changing the original class (Proxy) - You have many objects that share intrinsic state and can safely share it across instances (Flyweight) --- ## When NOT to Use Avoid this skill for: - Simple functions or methods that accomplish the task directly — do not introduce a pattern for pattern's sake (see YAGNI) - Inheritance hierarchies where subclassing cleanly covers the variation — Template Method may apply, but prefer Strategy for runtime swaps - When you can use composition and direct method calls without needing decoupled event notification — Observer adds unnecessary indirection - Performance-critical inner loops where the indirection overhead of a pattern matters — inline the logic directly - Domain models that already have clear responsibility boundaries — do not force patterns onto well-designed code (see SOLID as the primary guide) --- ## Core Workflow 1. **Identify the Structural vs. Behavioral Problem** — Determine whether the problem is about object composition/structure (Adapter, Bridge, Composite, Decorator, Facade, Proxy, Flyweight) or about object responsibilities and communication (Observer, Strategy, Command, Template Method, Mediator, Chain of Responsibility). **Checkpoint:** If the problem is "make these two interfaces work together," it is structural. If it is "change behavior at runtime" or "notify multiple objects," it is behavioral. 2. **Select the Pattern Based on the Specific Problem** — Map the problem to the correct pattern using the decision matrix: - Incompatible interface → Adapter - Need abstraction separated from implementation details → Bridge - Tree-like part-whole hierarchy → Composite - Add behaviors dynamically without subclassing → Decorator - Simplify complex subsystem API → Facade - Control/lazy-load/access-check an object → Proxy - Share data among many instances → Flyweight - Swap algorithms at runtime → Strategy - Encapsulate requests with undo/redo → Command - Define algorithm skeleton in a base class → Template Method - Decouple event publishers from subscribers → Observer - Centralize communication between many objects → Mediator - Process requests through multiple handlers sequentially → Chain of Responsibility 3. **Define the Interfaces (Protocols)** — Use `typing.Protocol` or `abc.ABC` to define the common interface that all concrete participants will implement. This is the critical step that ensures type safety and polymorphic substitution. **Checkpoint:** Every concrete class in the pattern must satisfy the protocol/ABC interface. Verify with static type checking before proceeding. 4. **Implement Participants with Composition** — Build each pattern participant as a separate class, connected through composition (aggregation) rather than inheritance where possible. Each class should have a single responsibility aligned with its role in the pattern. **Checkpoint:** No class should do more than what the GoF definition of its role requires. Apply SRP strictly. 5. **Wire Participants Together** — Connect the subject/client to the adaptee/implementation/strategy/command via dependency injection (constructor or setter). Ensure the client interacts only with the abstract interface, never with concrete implementations directly. **Checkpoint:** The client should compile and work if you swap any concrete participant for another that satisfies the same protocol. 6. **Add Error Handling and Edge Cases** — Implement guard clauses in pattern entry points (e.g., empty observer lists in Observer, null commands in Command). Validate state transitions where applicable (e.g., command state before undo). **Checkpoint:** Pattern runtime should never crash due to an empty collection, missing handler, or null reference. --- ## Implementation Patterns / Reference Guide ### Pattern 1: Adapter Pattern Makes incompatible interfaces work together by wrapping one interface in a new interface that clients expect. The adapter translates calls to the adaptee's native interface. Use Adapter when you must integrate a third-party library, legacy code, or an external service whose API does not match what your application expects. It is the only structural pattern that deals with interface incompatibility rather than object structure. ```python from __future__ import annotations import logging from typing import Protocol, runtime_checkable logger = logging.getLogger(__name__) @runtime_checkable class PaymentProcessor(Protocol): """The target interface expected by the application.""" def process_payment(self, amount: float, currency: str) -> bool: ... @runtime_checkable class LegacyPaymentGateway(Protocol): """The existing (incompatible) interface we need to adapt.""" def send_payment_request(self, value: int, code: str) -> dict[str, str]: ... class PaymentGatewayAdapter(PaymentProcessor): """Adapts LegacyPaymentGateway to the PaymentProcessor target interface. The adapter sits between the client (which expects PaymentProcessor) and the legacy system (which exposes LegacyPaymentGateway). It translates parameter names, units, and response formats. """ def __init__(self, gateway: LegacyPaymentGateway) -> None: self._gateway = gateway def process_payment(self, amount: float, currency: str) -> bool: try: # Convert from dollars (float) to cents (int), normalize currency code cents = int(round(amount * 100)) result = self._gateway.send_payment_request(cents, currency.upper()) if result.get("status") == "SUCCESS": logger.info("Payment of %.2f %s processed", amount, currency) return True else: logger.warning("Payment failed: %s", result.get("message", "unknown")) return False except Exception as exc: logger.error("Adapter error processing %.2f %s: %s", amount, currency, exc) raise RuntimeError(f"Payment adapter failed: {exc}") from exc # Example usage — client code knows only about PaymentProcessor def checkout(processor: PaymentProcessor, total: float, currency: str) -> None: success = processor.process_payment(total, currency) if not success: raise ValueError("Checkout failed") ``` **❌ BAD — Modifying the legacy class directly to match the new interface:** ```python # ❌ BAD: Violates Open/Closed Principle — modifying external/legacy code class BrokenLegacyGateway(LegacyPaymentGateway): def send_payment_request(self, value: int, code: str) -> dict[str, str]: # Someone had to modify this class just to fit our app... result = original_gateway.send_payment_request(value, code) if result.get("status") == "SUCCESS": return {"ok": True} # Breaking the contract of LegacyPaymentGateway return {"ok": False} ``` **✅ GOOD — Adapter wraps without modification:** ```python # ✅ GOOD: Legacy code untouched; adapter translates between worlds class CleanGatewayAdapter(PaymentProcessor): def __init__(self, gateway: LegacyPaymentGateway) -> None: self._gateway = gateway def process_payment(self, amount: float, currency: str) -> bool: cents = int(round(amount * 100)) result = self._gateway.send_payment_request(cents, currency.upper()) return result.get("status") == "SUCCESS" ``` **Practical note:** Use Adapter when integrating external libraries, migrating from one API version to another, or making a legacy module conform to a new interface contract. Choose Object Adapter (composition) over Class Adapter (inheritance) in Python — Python's multiple inheritance is fragile for this purpose. --- ### Pattern 2: Strategy Pattern Defines a family of interchangeable algorithms, encapsulates each one, and makes them swappable at runtime. The context holds a reference to a strategy interface and delegates to it instead of implementing the algorithm directly. Use Strategy when you have multiple variants of an algorithm (e.g., sorting orders, pricing models, notification delivery methods) and you need to switch between them based on configuration, user choice, or runtime conditions — without bloating a single class with conditional logic. ```python from __future__ import annotations import math from typing import Protocol class PricingStrategy(Protocol): """All pricing algorithms must implement this interface.""" def calculate_price(self, base_price: float, quantity: int) -> float: ... class FlatRatePricing: """Charges the standard list price per unit.""" def calculate_price(self, base_price: float, quantity: int) -> float: if quantity <= 0: raise ValueError("Quantity must be positive") return round(base_price * quantity, 2) class VolumeDiscountPricing: """Applies tiered volume discounts based on quantity thresholds.""" TIERS: list[tuple[int, float]] = [ (10, 0.05), # 10+ units → 5% off (50, 0.10), # 50+ units → 10% off (100, 0.15), # 100+ units → 15% off ] def calculate_price(self, base_price: float, quantity: int) -> float: if quantity <= 0: raise ValueError("Quantity must be positive") discount = 0.0 for threshold, rate in self.TIERS: if quantity >= threshold: discount = rate subtotal = base_price * quantity return round(subtotal * (1 - discount), 2) class DynamicPricing: """Adjusts price based on demand multiplier and time-of-day factor.""" def __init__(self, demand_multiplier: float = 1.0, time_factor: float = 1.0) -> None: self.demand_multiplier = max(0.5, min(demand_multiplier, 3.0)) self.time_factor = max(0.7, min(time_factor, 2.0)) def calculate_price(self, base_price: float, quantity: int) -> float: if quantity <= 0: raise ValueError("Quantity must be positive") adjusted = base_price * self.demand_multiplier * self.time_factor return round(adjusted * quantity, 2) class PricingContext: """Delegates pricing decisions to a strategy. Can swap strategies at runtime.""" def __init__(self, strategy: PricingStrategy) -> None: self._strategy = strategy @property def strategy(self) -> PricingStrategy: return self._strategy @strategy.setter def strategy(self, strategy: PricingStrategy) -> None: """Swap to a different pricing strategy at runtime.""" self._strategy = strategy def calculate(self, base_price: float, quantity: int) -> float: return self._strategy.calculate_price(base_price, quantity) # Example usage — context works with any strategy without knowing which one def order_total(context: PricingContext, price: float, qty: int) -> float: return context.calculate(price, qty) ``` **❌ BAD — Conditional logic instead of Strategy:** ```python # ❌ BAD: New pricing type requires modifying the class every time class BrokenOrderProcessor: def __init__(self, pricing_type: str = "flat") -> None: self._pricing_type = pricing_type def calculate(self, base_price: float, quantity: int) -> float: if self._pricing_type == "flat": return round(base_price * quantity, 2) elif self._pricing_type == "volume": discount = 0.10 if quantity >= 50 else 0.05 return round(base_price * quantity * (1 - discount), 2) elif self._pricing_type == "dynamic": demand = 1.5 # magic constant — should be configurable return round(base_price * demand * quantity, 2) else: raise ValueError(f"Unknown pricing type: {self._pricing_type}") ``` **✅ GOOD — Strategy pattern with interchangeable algorithms:** ```python # ✅ GOOD: New strategy = new class, no modification to context context = PricingContext(FlatRatePricing()) assert context.calculate(10.0, 5) == 50.0 context.strategy = VolumeDiscountPricing() assert context.calculate(10.0, 60) == 540.0 # 10% volume discount applied context.strategy = DynamicPricing(demand_multiplier=1.5) assert context.calculate(10.0, 5) == 75.0 ``` **Practical note:** Choose Strategy when behavior varies by configuration or external conditions and you want the selection logic decoupled from the execution logic. Use it instead of inheritance for runtime-swappable behavior; use inheritance (Template Method) for compile-time class hierarchies with shared step sequences. --- ### Pattern 3: Observer Pattern Establishes a one-to-many dependency between objects so that when one object (the subject) changes state, all its dependents (observers) are notified and updated automatically. Enables event-driven architecture with loose coupling between publishers and subscribers. Use Observer when multiple components need to react to the same state change without knowing about each other directly. It is the backbone of event systems, pub/sub messaging, reactive UI frameworks, and domain events in DDD. ```python from __future__ import annotations import weakref from typing import Protocol from abc import ABC class Observer(ABC): """Base class for all observers.""" def update(self, subject: str, event_data: dict) -> None: ... class Subject: """Maintains a list of observers and notifies them of state changes. Uses weak references to prevent memory leaks from forgotten unsubscription. Thread-safe for basic add/remove/notify operations via internal lock-free copy-on-read pattern for the observer list. """ def __init__(self) -> None: self._observers: list[weakref.ref[Observer]] = [] self._state: dict[str, object] = {} @property def state(self) -> dict[str, object]: return dict(self._state) @state.setter def state(self, value: dict[str, object]) -> None: """Update state and notify all registered observers.""" self._state = dict(value) # Defensive copy self._notify_all() def register(self, observer: Observer) -> None: if not isinstance(observer, Observer): raise TypeError("Observer must implement the Observer protocol") ref = weakref.ref(observer) if ref not in self._observers: self._observers.append(ref) def unregister(self, observer: Observer) -> None: target_ref = weakref.ref(observer) self._observers = [r for r in self._observers if r != target_ref] def _notify_all(self) -> None: """Notify all live observers, filtering out garbage-collected ones.""" live_observers: list[weakref.ref[Observer]] = [] for ref in self._observers: obs = ref() if obs is not None: try: obs.update( subject=type(self).__name__, event_data=dict(self._state), ) except Exception as exc: # One observer's failure should not break others import logging logging.getLogger(__name__).warning( "Observer %s failed: %s", obs, exc ) live_observers.append(ref) # Dead reference is silently removed on next notify cycle self._observers = live_observers class PriceAlertObserver(Observer): """Watches for price changes and logs alerts.""" def __init__(self, threshold: float) -> None: self.threshold = threshold def update(self, subject: str, event_data: dict) -> None: current = event_data.get("price", 0.0) if abs(current - self._prev_price) > self.threshold and hasattr(self, "_prev_price"): direction = "↑" if current > self._prev_price else "↓" print(f"[{subject}] Price {direction}: ${self._prev_price:.2f} → ${current:.2f}") self._prev_price = current class EmailNotifierObserver(Observer): """Sends email notifications on specific events.""" def __init__(self, email: str) -> None: self.email = email def update(self, subject: str, event_data: dict) -> None: if event_data.get("critical"): print(f"[{subject}] Email alert sent to {email}: " f"Critical event — data={event_data}") # Example usage — subject has no knowledge of concrete observer types def demonstrate_observer() -> None: market = Subject() # Register observers alert = PriceAlertObserver(threshold=0.50) notifier = EmailNotifierObserver(email="trader@example.com") market.register(alert) market.register(notifier) # State changes trigger automatic notifications market.state = {"price": 150.0, "critical": False} market.state = {"price": 149.30, "critical": False} market.state = {"price": 125.00, "critical": True} # Unregister to stop receiving updates market.unregister(notifier) ``` **❌ BAD — Tightly coupled state changes via direct calls:** ```python # ❌ BAD: Every change requires manually calling every dependent class BrokenPriceTracker: def __init__(self) -> None: self.alert_service = AlertService() self.email_service = EmailService() self.logger = Logger() self._price: float = 0.0 @property def price(self) -> float: return self._price @price.setter def price(self, value: float) -> None: self._price = value # Every new consumer adds another direct call here — violates OCP self.alert_service.check_threshold(self._price) self.email_service.send_alert(self._price) self.logger.log_price_change(self._price) ``` **✅ GOOD — Observer pattern with loose coupling:** ```python # ✅ GOOD: Adding a new observer requires no changes to the subject tracker = Subject() tracker.register(PriceAlertObserver(threshold=1.0)) tracker.register(EmailNotifierObserver("ops@example.com")) tracker.register(LoggerObserver()) # New observer — zero changes to Subject tracker.state = {"price": 99.99} ``` **Practical note:** Use Observer when you have a broadcast notification requirement and multiple consumers should react independently. Guard against memory leaks by using weak references in the subject's observer registry. In async/await systems, consider async observers with `asyncio.create_task` to avoid blocking the subject's notify cycle. --- ### Pattern 4: Command Pattern Encapsulates a request as an object, thereby letting you parameterize clients with different requests, queue or log requests, and support undo/redo operations. Each command object captures all information needed to execute the action. Use Command when you need to decouple the invoker (which triggers actions) from the receiver (which performs them), when you need to queue tasks for later execution, or when you need undo/redo capability — such as in editor applications, trading order systems, and workflow engines. ```python from __future__ import annotations from abc import ABC, abstractmethod from typing import Any class Command(ABC): """Abstract base for all commands.""" @abstractmethod def execute(self) -> str: ... @abstractmethod def undo(self) -> str: ... class OrderCommand(Command): """Encapsulates a trade order with full undo support. Stores the previous state so undo restores it exactly. Immutable command objects enable thread-safe command queues. """ def __init__(self, account: Account, symbol: str, quantity: int, side: str) -> None: self.account = account self.symbol = symbol self.quantity = quantity self.side = side # "BUY" or "SELL" def execute(self) -> str: if self.side == "BUY": result = self.account.buy(self.symbol, self.quantity) elif self.side == "SELL": result = self.account.sell(self.symbol, self.quantity) else: raise ValueError(f"Invalid side: {self.side}") return f"{self.side} {self.quantity} {self.symbol}: {result}" def undo(self) -> str: """Reverse the trade by doing the opposite operation.""" opposite = "SELL" if self.side == "BUY" else "BUY" return self.account.sell(self.symbol, self.quantity) \ if opposite == "SELL" else self.account.buy(self.symbol, self.quantity) class Account: """Simple account with buy/sell operations and state tracking for undo.""" def __init__(self, initial_cash: float = 10_000.0) -> None: self.cash = initial_cash self.positions: dict[str, int] = {} def buy(self, symbol: str, quantity: int) -> str: if quantity <= 0: raise ValueError("Quantity must be positive") price = 150.0 + hash(symbol) % 50 # Deterministic fake price cost = price * quantity if cost > self.cash: return f"Insufficient funds: need ${cost:.2f}, have ${self.cash:.2f}" self.cash -= cost self.positions[symbol] = self.positions.get(symbol, 0) + quantity return f"Purchased {quantity} {symbol} @ ${price:.2f}" def sell(self, symbol: str, quantity: int) -> str: if quantity <= 0: raise ValueError("Quantity must be positive") available = self.positions.get(symbol, 0) if available < quantity: return f"Insufficient {symbol} position: have {available}, want {quantity}" price = 155.0 + hash(symbol) % 45 # Deterministic fake price revenue = price * quantity self.cash += revenue self.positions[symbol] = available - quantity return f"Sold {quantity} {symbol} @ ${price:.2f}" class CommandInvoker: """Manages command execution with undo/redo history. Maintains separate stacks for executed and undone commands, enabling full undo/redo support. """ def __init__(self) -> None: self._history: list[Command] = [] self._undone: list[Command] = [] def execute_command(self, command: Command) -> str: result = command.execute() self._history.append(command) self._undone.clear() # New action invalidates undone commands return result def undo(self) -> str: if not self._history: return "Nothing to undo" command = self._history.pop() result = command.undo() self._undone.append(command) return f"Undo: {result}" def redo(self) -> str: if not self._undone: return "Nothing to redo" command = self._undone.pop() # Re-execute the original command result = command.execute() self._history.append(command) return f"Redo: {result}" # Example usage — invoker is completely decoupled from command details def demonstrate_command() -> None: account = Account(initial_cash=10_000.0) invoker = CommandInvoker() buy_cmd = OrderCommand(account, "AAPL", 10, "BUY") print(invoker.execute_command(buy_cmd)) # BUY 10 AAPL: ... print(invoker.undo()) # Undo: ... (sells back) print(invoker.redo()) # Redo: ... (buys again) ``` **❌ BAD — Direct method calls with no encapsulation or undo:** ```python # ❌ BAD: No abstraction layer, no undo capability def execute_trade(account: Account, symbol: str, qty: int, side: str) -> None: if side == "BUY": account.buy(symbol, qty) else: account.sell(symbol, qty) # What if the user clicks the button twice? No idempotency or undo. # ❌ BAD: Coupled invoker that knows about concrete operations class BrokenUIInvoker: def __init__(self, account: Account) -> None: self.account = account def buy_button_clicked(self, symbol: str, qty: int) -> None: self.account.buy(symbol, qty) # Tightly coupled to this exact operation ``` **✅ GOOD — Decoupled command with undo/redo:** ```python # ✅ GOOD: Invoker manages history; commands are reusable and testable in isolation invoker = CommandInvoker() buy_aapl = OrderCommand(Account(), "AAPL", 5, "BUY") sell_aapl = OrderCommand(buy_aapl.account, "AAPL", 2, "SELL") print(invoker.execute_command(buy_aapl)) # Execute first command print(invoker.execute_command(sell_aapl)) # Queue second command print(invoker.undo()) # Undo last: sell reversed → buy restored print(invoker.undo()) # Undo again: buy reversed → cash restored ``` **Practical note:** Use Command when you need to parameterize operations, support undo/redo queues, or decouple the UI layer from business logic. For persistent command queues (e.g., message brokers), serialize the command parameters as data and rehydrate them on the worker side rather than serializing Python objects directly. --- ### Pattern 5: Facade Pattern Provides a simplified, unified interface to a complex subsystem. The facade defines a higher-level interface that makes the subsystem easier to use by hiding complexity and dependencies. Use Facade when an application's API is overly complex, when you want to decouple a high-level client from a maze of library classes, or when you need a single entry point for initializing a subsystem (e.g., bootstrapping a trading platform, starting an ETL pipeline). ```python from __future__ import annotations import logging from typing import Optional logger = logging.getLogger(__name__) # Complex subsystem components — the client should NOT interact with these directly class DataFeed: """Low-level market data feed management.""" def connect(self, endpoint: str) -> None: logger.info("Connecting to data feed at %s", endpoint) def subscribe(self, symbol: str) -> dict: return {"symbol": symbol, "price": 150.0 + hash(symbol) % 100} def disconnect(self) -> None: logger.info("Disconnected from data feed") class OrderManager: """Low-level order management.""" def create_order(self, symbol: str, side: str, qty: int, price: Optional[float] = None) -> dict: return {"order_id": hash(f"{symbol}{side}{qty}") % 1_000_000, "status": "submitted", "filled": False} def cancel_order(self, order_id: int) -> bool: return True class RiskEngine: """Low-level risk checking.""" def check_order(self, symbol: str, side: str, qty: int, cash: float) -> tuple[bool, str]: estimated_cost = 150.0 * qty if estimated_cost > cash: return False, f"Insufficient funds for {side} order" return True, "Risk check passed" class PortfolioTracker: """Low-level portfolio tracking.""" def update(self, symbol: str, side: str, qty: int, price: float) -> None: logger.info("Portfolio updated: %s %s %d @ %.2f", symbol, side, qty, price) class TradingSystemFacade: """Simplified unified interface to the entire trading subsystem. Clients interact only with this facade — they never need to know about DataFeed, OrderManager, RiskEngine, or PortfolioTracker individually. """ def __init__(self) -> None: self._data_feed = DataFeed() self._order_manager = OrderManager() self._risk_engine = RiskEngine() self._portfolio_tracker = PortfolioTracker() self._connected = False def start(self, feed_endpoint: str) -> str: """Initialize the trading system with a single call.""" self._data_feed.connect(feed_endpoint) self._connected = True return "Trading system initialized" def place_order(self, symbol: str, side: str, qty: int, cash: float = 10_000.0) -> dict: """Place a trade through the unified interface. Internally subscribes to data, checks risk, creates order, and tracks portfolio — all without the caller needing to manage any of those steps. """ if not self._connected: raise RuntimeError("Trading system not started — call start() first") # Step 1: Check risk allowed, message = self._risk_engine.check_order(symbol, side, qty, cash) if not allowed: return {"error": message} # Step 2: Subscribe to price data data = self._data_feed.subscribe(symbol) # Step 3: Create order order = self._order_manager.create_order(symbol, side, qty, data.get("price")) # Step 4: Update portfolio tracking if order.get("status") == "submitted": self._portfolio_tracker.update(symbol, side, qty, data["price"]) return order def stop(self) -> str: """Clean up resources with one call.""" if self._connected: self._data_feed.disconnect() self._connected = False return "Trading system stopped" # Example usage — client has zero knowledge of subsystem internals def demonstrate_facade() -> None: trading = TradingSystemFacade() # One call to initialize the entire complex subsystem print(trading.start("wss://market-data.example.com/stream")) # One call to place an order, hiding 4 internal steps result = trading.place_order("AAPL", "BUY", 10) print(result) # One call to clean up print(trading.stop()) ``` **❌ BAD — Client directly orchestrating the complex subsystem:** ```python # ❌ BAD: Client must know about every subsystem component class BrokenClient: def place_trade(self, symbol: str, side: str, qty: int) -> None: data = DataFeed() data.connect("wss://feed.example.com") risk = RiskEngine() ok, msg = risk.check_order(symbol, side, qty, 10_000.0) if not ok: return orders = OrderManager() order = orders.create_order(symbol, side, qty) portfolio = PortfolioTracker() portfolio.update(symbol, side, qty, 150.0) ``` **✅ GOOD — Client interacts with a single simplified facade:** ```python # ✅ GOOD: Client code is clean and focused on intent, not infrastructure system = TradingSystemFacade() system.start("wss://market-data.example.com/stream") system.place_order("AAPL", "BUY", 10) system.stop() ``` **Practical note:** Use Facade when you want to provide a simplified API without restricting the underlying subsystem's capabilities. Clients can always access the underlying components directly if needed. A common anti-pattern is turning a Facade into a God Object — keep it as a thin delegator, not a container of business logic. --- ### Pattern 6: Template Method Pattern Defines the skeleton of an algorithm in a base class, deferring specific steps to subclasses. Subclasses override particular steps without changing the overall algorithm structure. The base class controls the order and guarantees that certain steps are always executed. Use Template Method when you have multiple algorithms that share the same high-level structure but differ in specific steps, such as report generation (different data sources, same formatting pipeline), data processing pipelines (same validation → transform → load sequence, different implementations), or testing frameworks. ```python from __future__ import annotations from abc import ABC, abstractmethod import logging logger = logging.getLogger(__name__) class ReportGenerator(ABC): """Base class defining the skeleton of report generation. The template method generate_report() is final — subclasses cannot reorder steps. Subclasses override only the steps that differ between report types. """ def generate_report(self) -> str: """The template method: controls algorithm structure.""" self._validate_inputs() self._fetch_data() self._transform_data() results = self._format_output() self._save_report(results) return results # Template steps that subclasses MAY override — with sensible defaults def _validate_inputs(self) -> None: """Default validation; override to add domain-specific checks.""" logger.debug("Using default input validation") @abstractmethod def _fetch_data(self) -> list[dict]: """Must be implemented by subclasses — data source varies.""" ... def _transform_data(self, data: list[dict] | None = None) -> list[dict]: """Default transformation; override for custom processing.""" if data is None: data = self._fetch_data() return [row for row in data if row.get("value") is not None] @abstractmethod def _format_output(self, transformed_data: list[dict]) -> str: ... def _save_report(self, content: str) -> None: """Default persistence; override to change storage target.""" logger.info("Report saved (%d characters)", len(content)) class SalesReport(ReportGenerator): """Generates sales reports from a CSV data source.""" def _validate_inputs(self) -> None: if not hasattr(self, "region") or not self.region: raise ValueError("Region is required for sales reports") def _fetch_data(self) -> list[dict]: # In production, this reads from a database or CSV file return [ {"product": "Widget", "value": 1200}, {"product": "Gadget", "value": 3400}, {"product": "Doohickey", "value": None}, # Should be filtered out ] def _format_output(self, transformed_data: list[dict]) -> str: lines = ["=== SALES REPORT ===", f"Region: {self.region}"] for row in transformed_data: lines.append(f" {row['product']}: ${row['value']}") return "\n".join(lines) def _save_report(self, content: str) -> None: logger.info("SALES REPORT saved to /reports/sales/%s", self.region) class PerformanceReport(ReportGenerator): """Generates performance reports from API metrics.""" def __init__(self, endpoint: str) -> None: self.endpoint = endpoint def _validate_inputs(self) -> None: if not self.endpoint.startswith(("http://", "https://")): raise ValueError(f"Invalid endpoint URL: {self.endpoint}") def _fetch_data(self) -> list[dict]: # In production, this calls the API return [ {"metric": "latency_p99", "value": 250}, {"metric": "error_rate", "value": 0.02}, {"metric": "throughput", "value": None}, # Filtered out ] def _format_output(self, transformed_data: list[dict]) -> str: lines = ["=== PERFORMANCE REPORT ==="] for row in transformed_data: lines.append(f" {row['metric']}: {row['value']}") return "\n".join(lines) # Example usage — the template method structure is fixed, only data differs def demonstrate_template_method() -> None: sales = SalesReport() sales.region = "North America" print(sales.generate_report()) perf = PerformanceReport(endpoint="https://api.example.com/metrics") print(perf.generate_report()) ``` **❌ BAD — Duplicate algorithm code in every subclass:** ```python # ❌ BAD: Every subclass reimplements the entire pipeline — violation of DRY class BrokenSalesReport: def generate(self) -> str: self._validate() # Redundant implementation data = self._fetch() # Different source, same structure cleaned = [d for d in data if d.get("value") is not None] # Copy-paste formatted = f"Sales:\n" + "\n".join(f" {v}" for v in cleaned) self._save(formatted) # Another copy of the save logic return formatted # Same validate, _save methods duplicated across every report type... class BrokenPerfReport: def generate(self) -> str: self._validate() # Nearly identical validation data = self._fetch() # Different source again cleaned = [d for d in data if d.get("value") is not None] # Exact copy formatted = f"Perf:\n" + "\n".join(f" {v}" for v in cleaned) self._save(formatted) # Same save logic again return formatted ``` **✅ GOOD — Template Method enforces structure once, varies only what differs:** ```python # ✅ GOOD: generate_report() is defined exactly once in the base class # Each subclass only overrides the steps that actually differ report = PerformanceReport(endpoint="https://metrics.internal/v1") print(report.generate_report()) # validate → fetch → transform → format → save ``` **Practical note:** Use Template Method when the algorithm structure is invariant but some steps vary. Use Strategy (behavioral pattern) instead when the entire algorithm varies and should be swappable at runtime — Template Method's variation points are locked at compile time via inheritance, while Strategy swaps entire algorithms at runtime. --- ### Pattern 7: Decorator Pattern Attaches additional responsibilities to an object dynamically. Decorators provide a flexible alternative to subclassing for extending functionality. The decorator wraps the original object and adds behavior before/after delegating to it. Use Decorator when you need to add behaviors (logging, caching, authent …(truncated)