Behavioral Patterns
The Patterns of Communication
Creational patterns ask how objects are born. Structural patterns ask how objects fit together. Behavioral patterns ask how objects talk to each other — how responsibility is distributed, how algorithms are encapsulated, and how state flows through a system.
Poor communication between objects is the root cause of the most painful codebases: tightly coupled components where every change ripples through the entire system, conditional logic that grows with every new requirement, and state machines implemented as chains of if/elif that no one dares to touch.
Behavioral patterns give you principled, named solutions to these problems — solutions that every senior engineer recognizes by name and can implement without thinking.
Quick Reference
| # | Pattern | Real-World Analogy | Used In |
|---|---|---|---|
| 1 | Observer | YouTube subscriptions | Event listeners, React useEffect, Kafka |
| 2 | Strategy | Choosing a route on Google Maps | Sorting algorithms, payment methods |
| 3 | Command | Restaurant order ticket | Undo/redo, job queues, Git commits |
| 4 | Iterator | Flipping through a book | for loops, Python generators, DB cursors |
| 5 | State | Traffic light transitions | Order status, TCP states, game AI |
The Five Behavioral Patterns
1. Observer
Intent: Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.
Also called: Pub/Sub, Event Listener, Signal/Slot.
When to use: UI event systems, real-time data feeds, model-view synchronization, notification pipelines.
from abc import ABC, abstractmethod
from typing import List, Any
class Observer(ABC):
@abstractmethod
def update(self, event: str, data: Any) -> None: pass
class Observable:
"""Base class for any object that can be observed."""
def __init__(self):
self._observers: dict[str, List[Observer]] = {}
def subscribe(self, event: str, observer: Observer) -> None:
self._observers.setdefault(event, []).append(observer)
def unsubscribe(self, event: str, observer: Observer) -> None:
if event in self._observers:
self._observers[event].remove(observer)
def notify(self, event: str, data: Any = None) -> None:
for observer in self._observers.get(event, []):
observer.update(event, data)
# Concrete Subject
class StockMarket(Observable):
def __init__(self):
super().__init__()
self._prices: dict[str, float] = {}
def update_price(self, symbol: str, price: float) -> None:
self._prices[symbol] = price
self.notify("price_change", {"symbol": symbol, "price": price})
# Concrete Observers
class PriceAlertBot(Observer):
def __init__(self, threshold: float):
self.threshold = threshold
def update(self, event: str, data: Any) -> None:
if data["price"] > self.threshold:
print(f"ALERT: {data['symbol']} hit ${data['price']:.2f}!")
class PriceLogger(Observer):
def update(self, event: str, data: Any) -> None:
print(f"LOG: {data['symbol']} = ${data['price']:.2f}")
market = StockMarket()
market.subscribe("price_change", PriceAlertBot(threshold=150.0))
market.subscribe("price_change", PriceLogger())
market.update_price("AAPL", 145.00) # logged only
market.update_price("AAPL", 155.00) # logged + alerted
// Observer in JavaScript — EventEmitter (Node.js built-in)
const EventEmitter = require("events");
class OrderService extends EventEmitter {
placeOrder(order) {
// ... process order ...
this.emit("order.placed", order); // notify all listeners
}
}
const orders = new OrderService();
// Multiple independent listeners — decoupled from OrderService
orders.on("order.placed", (order) => console.log(`Email sent for order ${order.id}`));
orders.on("order.placed", (order) => console.log(`Inventory updated for ${order.id}`));
orders.on("order.placed", (order) => console.log(`Analytics recorded for ${order.id}`));
orders.placeOrder({ id: "ORD-001", item: "Laptop", amount: 999 });
2. Strategy
Intent: Define a family of algorithms, encapsulate each one, and make them interchangeable. Strategy lets the algorithm vary independently from clients that use it.
When to use: When you have multiple algorithms for a task and want to switch between them at runtime — sorting strategies, compression algorithms, payment methods, routing strategies.
from abc import ABC, abstractmethod
from typing import List
class SortStrategy(ABC):
@abstractmethod
def sort(self, data: List[int]) -> List[int]: pass
class BubbleSort(SortStrategy):
def sort(self, data: List[int]) -> List[int]:
arr = data.copy()
n = len(arr)
for i in range(n):
for j in range(n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j]
return arr
class QuickSort(SortStrategy):
def sort(self, data: List[int]) -> List[int]:
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
mid = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return self.sort(left) + mid + self.sort(right)
class TimSort(SortStrategy):
def sort(self, data: List[int]) -> List[int]:
return sorted(data) # Python's built-in Timsort
class Sorter:
"""Context — uses a strategy, can switch at runtime."""
def __init__(self, strategy: SortStrategy):
self._strategy = strategy
def set_strategy(self, strategy: SortStrategy) -> None:
self._strategy = strategy
def sort(self, data: List[int]) -> List[int]:
return self._strategy.sort(data)
data = [64, 34, 25, 12, 22, 11, 90]
sorter = Sorter(QuickSort())
print(sorter.sort(data)) # quicksort
sorter.set_strategy(TimSort())
print(sorter.sort(data)) # switch to Timsort at runtime
// Strategy for shipping cost calculation
const strategies = {
standard: (weight) => 5.99 + weight * 0.5,
express: (weight) => 12.99 + weight * 1.0,
overnight:(weight) => 24.99 + weight * 2.0,
};
class ShippingCalculator {
constructor(strategy) { this.strategy = strategy; }
setStrategy(strategy) { this.strategy = strategy; }
calculate(weight) { return this.strategy(weight); }
}
const calc = new ShippingCalculator(strategies.standard);
console.log(calc.calculate(2.5)); // $7.24
calc.setStrategy(strategies.express);
console.log(calc.calculate(2.5)); // $15.49
Strategy vs if/elif: Every if type == "X" or elif algorithm == "Y" that selects different behavior is a candidate for Strategy. When a new case is added, Strategy requires only a new class — no modification of existing code (OCP).
3. Command
Intent: Encapsulate a request as an object, thereby allowing parameterization of clients with different requests, queuing of requests, logging, and undoable operations.
When to use: Undo/redo systems, task queues, transaction logs, macro recording, request scheduling.
from abc import ABC, abstractmethod
from typing import List
class Command(ABC):
@abstractmethod
def execute(self) -> None: pass
@abstractmethod
def undo(self) -> None: pass
# Receiver — the object that knows how to perform the actual work
class TextEditor:
def __init__(self):
self.content = ""
def insert(self, text: str, position: int) -> None:
self.content = self.content[:position] + text + self.content[position:]
def delete(self, position: int, length: int) -> None:
self.content = self.content[:position] + self.content[position + length:]
# Concrete Commands
class InsertCommand(Command):
def __init__(self, editor: TextEditor, text: str, position: int):
self._editor = editor
self._text = text
self._position = position
def execute(self) -> None:
self._editor.insert(self._text, self._position)
def undo(self) -> None:
self._editor.delete(self._position, len(self._text))
class DeleteCommand(Command):
def __init__(self, editor: TextEditor, position: int, length: int):
self._editor = editor
self._position = position
self._length = length
self._deleted_text = ""
def execute(self) -> None:
self._deleted_text = self._editor.content[self._position:self._position + self._length]
self._editor.delete(self._position, self._length)
def undo(self) -> None:
self._editor.insert(self._deleted_text, self._position)
# Invoker — manages command history and undo/redo
class CommandHistory:
def __init__(self):
self._history: List[Command] = []
def execute(self, command: Command) -> None:
command.execute()
self._history.append(command)
def undo(self) -> None:
if self._history:
self._history.pop().undo()
editor = TextEditor()
history = CommandHistory()
history.execute(InsertCommand(editor, "Hello", 0))
history.execute(InsertCommand(editor, " World", 5))
print(editor.content) # Hello World
history.undo()
print(editor.content) # Hello
history.undo()
print(editor.content) # (empty)
4. Iterator
Intent: Provide a way to access elements of a collection sequentially without exposing its underlying representation.
When to use: Custom data structures that need to be iterable, traversal algorithms that should be decoupled from collection types.
from typing import Generic, TypeVar, Iterator as Iter
T = TypeVar("T")
class TreeNode:
def __init__(self, value: int):
self.value = value
self.left: TreeNode | None = None
self.right: TreeNode | None = None
class InOrderIterator:
"""Iterates a binary tree in-order (left, root, right) without recursion."""
def __init__(self, root: TreeNode | None):
self._stack: list[TreeNode] = []
self._push_left(root)
def _push_left(self, node: TreeNode | None) -> None:
while node:
self._stack.append(node)
node = node.left
def __iter__(self) -> "InOrderIterator":
return self
def __next__(self) -> int:
if not self._stack:
raise StopIteration
node = self._stack.pop()
self._push_left(node.right)
return node.value
# Build a BST
root = TreeNode(5)
root.left = TreeNode(3)
root.right = TreeNode(7)
root.left.left = TreeNode(1)
root.left.right = TreeNode(4)
# Use Python's for loop — works because we implement __iter__ and __next__
for value in InOrderIterator(root):
print(value, end=" ") # 1 3 4 5 7 — sorted order
Python's iteration protocol (__iter__, __next__) is the Iterator pattern built into the language. Every for loop, list comprehension, and in check uses it. Understanding the pattern explains how Python iteration works under the hood.
5. State
Intent: Allow an object to alter its behavior when its internal state changes. The object will appear to change its class.
When to use: Traffic lights, order status machines, vending machines, connection states, game character states — anything with well-defined states and transitions.
from abc import ABC, abstractmethod
class OrderState(ABC):
@abstractmethod
def confirm(self, order: "Order") -> None: pass
@abstractmethod
def ship(self, order: "Order") -> None: pass
@abstractmethod
def deliver(self, order: "Order") -> None: pass
@abstractmethod
def cancel(self, order: "Order") -> None: pass
class PendingState(OrderState):
def confirm(self, order: "Order") -> None:
print("Order confirmed.")
order.set_state(ConfirmedState())
def ship(self, order: "Order") -> None:
print("Cannot ship — order not confirmed yet.")
def deliver(self, order: "Order") -> None:
print("Cannot deliver — order not confirmed yet.")
def cancel(self, order: "Order") -> None:
print("Order cancelled.")
order.set_state(CancelledState())
class ConfirmedState(OrderState):
def confirm(self, order: "Order") -> None:
print("Order already confirmed.")
def ship(self, order: "Order") -> None:
print("Order shipped.")
order.set_state(ShippedState())
def deliver(self, order: "Order") -> None:
print("Cannot deliver — order not shipped yet.")
def cancel(self, order: "Order") -> None:
print("Order cancelled.")
order.set_state(CancelledState())
class ShippedState(OrderState):
def confirm(self, order: "Order") -> None:
print("Order already confirmed.")
def ship(self, order: "Order") -> None:
print("Order already shipped.")
def deliver(self, order: "Order") -> None:
print("Order delivered!")
order.set_state(DeliveredState())
def cancel(self, order: "Order") -> None:
print("Cannot cancel — order already shipped.")
class DeliveredState(OrderState):
def confirm(self, order): print("Order already delivered.")
def ship(self, order): print("Order already delivered.")
def deliver(self, order): print("Order already delivered.")
def cancel(self, order): print("Cannot cancel — order delivered.")
class CancelledState(OrderState):
def confirm(self, order): print("Cannot confirm — order cancelled.")
def ship(self, order): print("Cannot ship — order cancelled.")
def deliver(self, order): print("Cannot deliver — order cancelled.")
def cancel(self, order): print("Order already cancelled.")
class Order:
def __init__(self):
self._state: OrderState = PendingState()
def set_state(self, state: OrderState) -> None:
self._state = state
def confirm(self) -> None: self._state.confirm(self)
def ship(self) -> None: self._state.ship(self)
def deliver(self) -> None: self._state.deliver(self)
def cancel(self) -> None: self._state.cancel(self)
order = Order()
order.ship() # Cannot ship — order not confirmed yet.
order.confirm() # Order confirmed.
order.ship() # Order shipped.
order.cancel() # Cannot cancel — order already shipped.
order.deliver() # Order delivered!
State vs if/elif: A state machine implemented with if self.status == "pending" becomes unmaintainable as states grow. The State pattern distributes the logic across classes — each state class knows only what transitions it allows, making the code open for extension (new state = new class) without modifying existing states.
Pattern Comparison
| Pattern | Problem Solved | Key Idea |
|---|---|---|
| Observer | Notify multiple dependents of changes | Subscribe/notify — loose coupling |
| Strategy | Swap algorithms at runtime | Encapsulate algorithm in a class |
| Command | Undoable, queueable requests | Encapsulate request as an object |
| Iterator | Traverse collection without exposing internals | Unified traversal interface |
| State | Different behavior per state | Delegate behavior to state object |
Real-World Examples
- Observer: React's
useEffectdependency array, Reduxstore.subscribe(), DOMaddEventListener, Kafka consumer groups, WebSocket event handlers. - Strategy: Python's
sorted(key=...)accepts any comparison strategy. Payment processors (Stripe, PayPal, Crypto) implement the same charge interface. A/B testing selects ranking strategies per user. - Command: Git commits are commands — each is a recorded change that can be reverted. Database transactions. Browser history (back/forward). Queue systems (Celery tasks).
- Iterator: Python's
forloop, generators (yield), database cursors, Kafka consumer offset traversal, filereadline(). - State: TCP connection states (LISTEN → SYN_RECEIVED → ESTABLISHED → CLOSE_WAIT). Redux reducers transition app state. Game character AI (idle → patrolling → attacking → fleeing).
Common Mistakes
Observer creating memory leaks. If observers register but never unregister, the subject holds references preventing garbage collection. Always provide and call unsubscribe(). In JavaScript, use removeEventListener or AbortController with addEventListener.
Strategy adding indirection without benefit. If you only ever have one algorithm and no plan to add others, Strategy adds classes without value. Apply it when you have (or anticipate) two or more interchangeable algorithms.
Command not saving enough state for undo. A delete command that deletes but does not save what was deleted cannot implement undo. Before execute(), always snapshot the state needed by undo().
State explosion. If your system has 20 states and 15 possible transitions, the State pattern creates 20 classes. Consider a state machine table (dictionary of transitions) for simpler cases — it is less object-oriented but far less verbose.
Summary
- Observer: One-to-many notification. Subjects notify all subscribers when state changes — without knowing who they are.
- Strategy: Encapsulate interchangeable algorithms. Swap behavior at runtime without changing the context class.
- Command: Turn requests into objects. Enables undo/redo, queuing, logging, and macro recording.
- Iterator: Uniform traversal interface for any collection. Python's
forloop and__iter__/__next__protocol implement this natively. - State: Encapsulate state-specific behavior in state classes. Eliminates state-checking conditionals and makes transitions explicit.
- Behavioral patterns focus on communication and responsibility distribution — who calls whom, and who decides what.