1"""
2Behavioral Design Patterns
3
4Patterns for communication between objects:
51. Observer - Define subscription mechanism to notify multiple objects
62. Strategy - Define family of algorithms, make them interchangeable
73. Command - Encapsulate request as object
8"""
9
10from abc import ABC, abstractmethod
11from typing import List, Dict, Any, Callable, Optional
12from dataclasses import dataclass
13from datetime import datetime
14from enum import Enum
15
16
17# =============================================================================
18# 1. OBSERVER PATTERN
19# Define one-to-many dependency so when one object changes state,
20# all dependents are notified
21# =============================================================================
22
23print("=" * 70)
24print("1. OBSERVER PATTERN")
25print("=" * 70)
26
27
28# Observer interface
29class Observer(ABC):
30 """Interface for observers that watch subjects"""
31
32 @abstractmethod
33 def update(self, subject: 'Subject') -> None:
34 """Called when subject's state changes"""
35 pass
36
37
38# Subject (Observable)
39class Subject:
40 """
41 Maintains list of observers and notifies them of state changes.
42 """
43
44 def __init__(self):
45 self._observers: List[Observer] = []
46 self._state: Any = None
47
48 def attach(self, observer: Observer) -> None:
49 """Add an observer"""
50 if observer not in self._observers:
51 self._observers.append(observer)
52 print(f"Attached observer: {observer.__class__.__name__}")
53
54 def detach(self, observer: Observer) -> None:
55 """Remove an observer"""
56 if observer in self._observers:
57 self._observers.remove(observer)
58 print(f"Detached observer: {observer.__class__.__name__}")
59
60 def notify(self) -> None:
61 """Notify all observers of state change"""
62 for observer in self._observers:
63 observer.update(self)
64
65 @property
66 def state(self) -> Any:
67 return self._state
68
69 @state.setter
70 def state(self, value: Any) -> None:
71 self._state = value
72 self.notify()
73
74
75# Concrete Subject - Stock
76@dataclass
77class Stock:
78 """Stock price that observers watch"""
79 symbol: str
80 price: float
81
82
83class StockMarket(Subject):
84 """Observable stock market"""
85
86 def __init__(self):
87 super().__init__()
88 self.stocks: Dict[str, Stock] = {}
89
90 def update_stock(self, symbol: str, price: float) -> None:
91 """Update stock price and notify observers"""
92 old_price = self.stocks.get(symbol).price if symbol in self.stocks else 0
93 self.stocks[symbol] = Stock(symbol, price)
94 self._state = (symbol, old_price, price)
95 self.notify()
96
97
98# Concrete Observers
99class EmailAlert(Observer):
100 """Observer that sends email alerts"""
101
102 def __init__(self, threshold: float):
103 self.threshold = threshold
104
105 def update(self, subject: Subject) -> None:
106 symbol, old_price, new_price = subject.state
107 change = abs(new_price - old_price)
108
109 if change >= self.threshold:
110 print(f"[EMAIL] {symbol}: ${old_price:.2f} → ${new_price:.2f} "
111 f"(change: ${change:.2f})")
112
113
114class SMSAlert(Observer):
115 """Observer that sends SMS alerts"""
116
117 def update(self, subject: Subject) -> None:
118 symbol, old_price, new_price = subject.state
119 if new_price > old_price:
120 print(f"[SMS] {symbol} UP: ${new_price:.2f} (+{new_price - old_price:.2f})")
121 elif new_price < old_price:
122 print(f"[SMS] {symbol} DOWN: ${new_price:.2f} (-{old_price - new_price:.2f})")
123
124
125class Logger(Observer):
126 """Observer that logs all changes"""
127
128 def update(self, subject: Subject) -> None:
129 symbol, old_price, new_price = subject.state
130 timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
131 print(f"[LOG] {timestamp} | {symbol}: ${new_price:.2f}")
132
133
134# =============================================================================
135# 2. STRATEGY PATTERN
136# Define family of algorithms, encapsulate each one, make them interchangeable
137# =============================================================================
138
139print("\n" + "=" * 70)
140print("2. STRATEGY PATTERN")
141print("=" * 70)
142
143
144# Strategy interface
145class PaymentStrategy(ABC):
146 """Abstract payment strategy"""
147
148 @abstractmethod
149 def pay(self, amount: float) -> str:
150 pass
151
152
153# Concrete strategies
154class CreditCardPayment(PaymentStrategy):
155 def __init__(self, card_number: str, cvv: str):
156 self.card_number = card_number
157 self.cvv = cvv
158
159 def pay(self, amount: float) -> str:
160 # Mask card number
161 masked = f"****-****-****-{self.card_number[-4:]}"
162 return f"Paid ${amount:.2f} with Credit Card {masked}"
163
164
165class PayPalPayment(PaymentStrategy):
166 def __init__(self, email: str):
167 self.email = email
168
169 def pay(self, amount: float) -> str:
170 return f"Paid ${amount:.2f} via PayPal ({self.email})"
171
172
173class CryptoPayment(PaymentStrategy):
174 def __init__(self, wallet_address: str, currency: str):
175 self.wallet_address = wallet_address
176 self.currency = currency
177
178 def pay(self, amount: float) -> str:
179 masked = f"{self.wallet_address[:6]}...{self.wallet_address[-4:]}"
180 return f"Paid ${amount:.2f} in {self.currency} to {masked}"
181
182
183class BankTransferPayment(PaymentStrategy):
184 def __init__(self, account_number: str):
185 self.account_number = account_number
186
187 def pay(self, amount: float) -> str:
188 masked = f"****{self.account_number[-4:]}"
189 return f"Paid ${amount:.2f} via Bank Transfer to {masked}"
190
191
192# Context that uses strategy
193class ShoppingCart:
194 """
195 Shopping cart that can use different payment strategies.
196 Strategy can be changed at runtime!
197 """
198
199 def __init__(self):
200 self.items: List[tuple[str, float]] = []
201 self._payment_strategy: Optional[PaymentStrategy] = None
202
203 def add_item(self, name: str, price: float):
204 self.items.append((name, price))
205
206 def calculate_total(self) -> float:
207 return sum(price for _, price in self.items)
208
209 def set_payment_strategy(self, strategy: PaymentStrategy):
210 """Change payment strategy at runtime"""
211 self._payment_strategy = strategy
212
213 def checkout(self) -> str:
214 if not self._payment_strategy:
215 return "Error: No payment method selected"
216
217 total = self.calculate_total()
218 return self._payment_strategy.pay(total)
219
220
221# Another example: Sorting strategies
222class SortStrategy(ABC):
223 @abstractmethod
224 def sort(self, data: List[int]) -> List[int]:
225 pass
226
227
228class QuickSort(SortStrategy):
229 def sort(self, data: List[int]) -> List[int]:
230 if len(data) <= 1:
231 return data
232 pivot = data[len(data) // 2]
233 left = [x for x in data if x < pivot]
234 middle = [x for x in data if x == pivot]
235 right = [x for x in data if x > pivot]
236 return self.sort(left) + middle + self.sort(right)
237
238
239class BubbleSort(SortStrategy):
240 def sort(self, data: List[int]) -> List[int]:
241 arr = data.copy()
242 n = len(arr)
243 for i in range(n):
244 for j in range(0, n - i - 1):
245 if arr[j] > arr[j + 1]:
246 arr[j], arr[j + 1] = arr[j + 1], arr[j]
247 return arr
248
249
250class SortContext:
251 """Context that uses sorting strategy"""
252
253 def __init__(self, strategy: SortStrategy):
254 self._strategy = strategy
255
256 def set_strategy(self, strategy: SortStrategy):
257 self._strategy = strategy
258
259 def sort_data(self, data: List[int]) -> List[int]:
260 return self._strategy.sort(data)
261
262
263# =============================================================================
264# 3. COMMAND PATTERN
265# Encapsulate request as object, parameterize clients with different requests
266# =============================================================================
267
268print("\n" + "=" * 70)
269print("3. COMMAND PATTERN")
270print("=" * 70)
271
272
273# Receiver - knows how to perform operations
274class TextEditor:
275 """Receiver that performs actual operations"""
276
277 def __init__(self):
278 self.text = ""
279
280 def insert_text(self, text: str, position: int):
281 self.text = self.text[:position] + text + self.text[position:]
282 print(f"Inserted '{text}' at position {position}")
283
284 def delete_text(self, start: int, length: int):
285 deleted = self.text[start:start + length]
286 self.text = self.text[:start] + self.text[start + length:]
287 print(f"Deleted '{deleted}' from position {start}")
288 return deleted
289
290 def get_text(self) -> str:
291 return self.text
292
293
294# Command interface
295class Command(ABC):
296 """Abstract command"""
297
298 @abstractmethod
299 def execute(self) -> None:
300 pass
301
302 @abstractmethod
303 def undo(self) -> None:
304 pass
305
306
307# Concrete commands
308class InsertTextCommand(Command):
309 """Command to insert text"""
310
311 def __init__(self, editor: TextEditor, text: str, position: int):
312 self.editor = editor
313 self.text = text
314 self.position = position
315
316 def execute(self) -> None:
317 self.editor.insert_text(self.text, self.position)
318
319 def undo(self) -> None:
320 self.editor.delete_text(self.position, len(self.text))
321
322
323class DeleteTextCommand(Command):
324 """Command to delete text"""
325
326 def __init__(self, editor: TextEditor, start: int, length: int):
327 self.editor = editor
328 self.start = start
329 self.length = length
330 self.deleted_text = ""
331
332 def execute(self) -> None:
333 self.deleted_text = self.editor.delete_text(self.start, self.length)
334
335 def undo(self) -> None:
336 self.editor.insert_text(self.deleted_text, self.start)
337
338
339# Invoker - manages command history
340class CommandHistory:
341 """Invoker that tracks command history for undo/redo"""
342
343 def __init__(self):
344 self.history: List[Command] = []
345 self.redo_stack: List[Command] = []
346
347 def execute_command(self, command: Command):
348 """Execute command and add to history"""
349 command.execute()
350 self.history.append(command)
351 self.redo_stack.clear() # Clear redo stack on new command
352
353 def undo(self):
354 """Undo last command"""
355 if not self.history:
356 print("Nothing to undo")
357 return
358
359 command = self.history.pop()
360 command.undo()
361 self.redo_stack.append(command)
362 print("Undone")
363
364 def redo(self):
365 """Redo last undone command"""
366 if not self.redo_stack:
367 print("Nothing to redo")
368 return
369
370 command = self.redo_stack.pop()
371 command.execute()
372 self.history.append(command)
373 print("Redone")
374
375
376# Another example: Smart home automation
377class Light:
378 """Receiver for light commands"""
379
380 def __init__(self, location: str):
381 self.location = location
382 self.is_on = False
383 self.brightness = 0
384
385 def turn_on(self):
386 self.is_on = True
387 self.brightness = 100
388 print(f"{self.location} light turned ON")
389
390 def turn_off(self):
391 self.is_on = False
392 self.brightness = 0
393 print(f"{self.location} light turned OFF")
394
395 def set_brightness(self, level: int):
396 if self.is_on:
397 self.brightness = level
398 print(f"{self.location} light brightness set to {level}%")
399
400
401class LightOnCommand(Command):
402 def __init__(self, light: Light):
403 self.light = light
404
405 def execute(self):
406 self.light.turn_on()
407
408 def undo(self):
409 self.light.turn_off()
410
411
412class LightOffCommand(Command):
413 def __init__(self, light: Light):
414 self.light = light
415
416 def execute(self):
417 self.light.turn_off()
418
419 def undo(self):
420 self.light.turn_on()
421
422
423class RemoteControl:
424 """Invoker - remote control with programmable buttons"""
425
426 def __init__(self):
427 self.commands: Dict[str, Command] = {}
428
429 def set_command(self, button: str, command: Command):
430 self.commands[button] = command
431
432 def press_button(self, button: str):
433 if button in self.commands:
434 self.commands[button].execute()
435 else:
436 print(f"Button '{button}' not programmed")
437
438
439# =============================================================================
440# DEMONSTRATIONS
441# =============================================================================
442
443def demonstrate_observer():
444 print("\n[OBSERVER DEMONSTRATION]")
445 print("-" * 50)
446
447 # Create subject
448 market = StockMarket()
449
450 # Create observers
451 email = EmailAlert(threshold=5.0)
452 sms = SMSAlert()
453 logger = Logger()
454
455 # Attach observers
456 market.attach(email)
457 market.attach(sms)
458 market.attach(logger)
459
460 # Update stocks - all observers notified
461 print("\nStock updates:")
462 market.update_stock("AAPL", 150.00)
463 market.update_stock("AAPL", 155.50) # Big change
464 market.update_stock("GOOGL", 2800.00)
465
466 # Detach email observer
467 print("\nDetaching email alerts...")
468 market.detach(email)
469
470 print("\nMore updates (no email):")
471 market.update_stock("AAPL", 160.00)
472
473 print("\nUse cases:")
474 print(" • Event handling systems")
475 print(" • MVC architecture")
476 print(" • Real-time data feeds")
477 print(" • Notification systems")
478
479
480def demonstrate_strategy():
481 print("\n[STRATEGY DEMONSTRATION]")
482 print("-" * 50)
483
484 # Create shopping cart
485 cart = ShoppingCart()
486 cart.add_item("Laptop", 1200.00)
487 cart.add_item("Mouse", 25.00)
488 cart.add_item("Keyboard", 75.00)
489
490 print(f"Cart total: ${cart.calculate_total():.2f}\n")
491
492 # Try different payment strategies
493 print("Payment with Credit Card:")
494 cart.set_payment_strategy(CreditCardPayment("1234567890123456", "123"))
495 print(cart.checkout())
496
497 print("\nPayment with PayPal:")
498 cart.set_payment_strategy(PayPalPayment("user@example.com"))
499 print(cart.checkout())
500
501 print("\nPayment with Crypto:")
502 cart.set_payment_strategy(CryptoPayment("0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb", "BTC"))
503 print(cart.checkout())
504
505 # Sorting example
506 print("\n\nSorting Strategy:")
507 data = [64, 34, 25, 12, 22, 11, 90]
508
509 context = SortContext(QuickSort())
510 print(f"QuickSort: {context.sort_data(data)}")
511
512 context.set_strategy(BubbleSort())
513 print(f"BubbleSort: {context.sort_data(data)}")
514
515 print("\nUse cases:")
516 print(" • Interchangeable algorithms")
517 print(" • Payment processing")
518 print(" • Data validation")
519 print(" • Compression algorithms")
520
521
522def demonstrate_command():
523 print("\n[COMMAND DEMONSTRATION]")
524 print("-" * 50)
525
526 # Text editor with undo/redo
527 editor = TextEditor()
528 history = CommandHistory()
529
530 print("Building text with commands:\n")
531
532 # Execute commands
533 history.execute_command(InsertTextCommand(editor, "Hello", 0))
534 print(f"Text: '{editor.get_text()}'\n")
535
536 history.execute_command(InsertTextCommand(editor, " World", 5))
537 print(f"Text: '{editor.get_text()}'\n")
538
539 history.execute_command(InsertTextCommand(editor, "!", 11))
540 print(f"Text: '{editor.get_text()}'\n")
541
542 # Undo operations
543 print("Undoing operations:")
544 history.undo()
545 print(f"Text: '{editor.get_text()}'\n")
546
547 history.undo()
548 print(f"Text: '{editor.get_text()}'\n")
549
550 # Redo
551 print("Redoing:")
552 history.redo()
553 print(f"Text: '{editor.get_text()}'\n")
554
555 # Smart home example
556 print("\nSmart Home Remote Control:")
557 living_room = Light("Living Room")
558 bedroom = Light("Bedroom")
559
560 remote = RemoteControl()
561 remote.set_command("1", LightOnCommand(living_room))
562 remote.set_command("2", LightOffCommand(living_room))
563 remote.set_command("3", LightOnCommand(bedroom))
564 remote.set_command("4", LightOffCommand(bedroom))
565
566 print("\nPressing buttons:")
567 remote.press_button("1")
568 remote.press_button("3")
569 remote.press_button("2")
570
571 print("\nUse cases:")
572 print(" • Undo/redo functionality")
573 print(" • Transaction systems")
574 print(" • Job queues")
575 print(" • GUI buttons/menu items")
576 print(" • Macro recording")
577
578
579def print_summary():
580 print("\n" + "=" * 70)
581 print("BEHAVIORAL PATTERNS SUMMARY")
582 print("=" * 70)
583
584 print("""
585OBSERVER
586 Purpose: Notify multiple objects of state changes
587 When: One-to-many relationships, event systems
588 Benefits: Loose coupling, dynamic subscriptions
589 Drawbacks: Can cause memory leaks, notification order issues
590
591STRATEGY
592 Purpose: Encapsulate algorithms, make them interchangeable
593 When: Multiple algorithms for same task
594 Benefits: Runtime algorithm selection, eliminates conditionals
595 Drawbacks: Clients must know strategies, more objects
596
597COMMAND
598 Purpose: Encapsulate requests as objects
599 When: Need undo/redo, queue operations, log requests
600 Benefits: Decouples sender/receiver, supports undo
601 Drawbacks: Many small classes, complexity
602
603CHOOSING THE RIGHT PATTERN:
604 • Observer: Multiple objects need updates
605 • Strategy: Need to switch algorithms at runtime
606 • Command: Need undo/redo or operation queuing
607
608REAL-WORLD EXAMPLES:
609 • Observer: Event listeners, data binding, pub-sub
610 • Strategy: Payment methods, sorting, validation
611 • Command: Text editors, transactions, smart home
612""")
613
614
615if __name__ == "__main__":
616 demonstrate_observer()
617 demonstrate_strategy()
618 demonstrate_command()
619 print_summary()