AiTechWorlds
AiTechWorlds
You're traveling in Europe with an American plug. The socket is a different shape. The device is perfectly good — it just can't connect. An electrical adapter solves this without modifying the plug or the socket. That's the Adapter pattern.
You run a newsletter. You write one article. Ten thousand subscribers are notified automatically. You didn't call each subscriber individually — the system broadcast the change. That's the Observer pattern.
Design patterns are proven solutions to recurring problems. Structural patterns deal with how classes and objects are composed. Behavioral patterns deal with how objects communicate and delegate responsibility.
The problem: You have a third-party library or legacy system with an incompatible interface. You cannot modify it, but you need it to work with your code.
The solution: Write an Adapter class that wraps the incompatible class and exposes the interface your code expects.
# --- The third-party payment library (you cannot change this) ---
class StripePaymentAPI:
"""External Stripe library — incompatible interface."""
def create_charge(self, amount_cents: int, currency: str, card_token: str) -> dict:
# Stripe uses cents, not dollars
return {
"id": "ch_abc123",
"amount": amount_cents,
"currency": currency,
"status": "succeeded"
}
# --- Your application's expected interface ---
class PaymentProcessor:
"""Interface your app uses for all payment processors."""
def charge(self, amount_dollars: float, card_token: str) -> bool:
raise NotImplementedError
# --- The Adapter bridges the gap ---
class StripeAdapter(PaymentProcessor):
def __init__(self, stripe_api: StripePaymentAPI):
self._stripe = stripe_api # Hold a reference to the incompatible object
def charge(self, amount_dollars: float, card_token: str) -> bool:
# Convert dollars to cents (Stripe's required format)
amount_cents = int(amount_dollars * 100)
# Translate to Stripe's method signature
result = self._stripe.create_charge(amount_cents, "usd", card_token)
return result["status"] == "succeeded" # Translate result to bool
Usage:
stripe_api = StripePaymentAPI() # The incompatible third-party object
processor = StripeAdapter(stripe_api) # Wrap it in the adapter
success = processor.charge(49.99, "tok_visa_test") # Clean, unified interface
print(success) # True
Swapping from Stripe to PayPal later only requires writing a PayPalAdapter — the rest of your app is untouched.
@decorator Syntax)The problem: You want to add behavior (logging, caching, authentication) to an object without modifying its class or creating a subclass for every combination.
The solution: Wrap the object in another object that adds the extra behavior, then delegates the core work to the original.
class DataService:
"""Core service — fetches data from a source."""
def fetch(self, query: str) -> str:
print(f" Fetching data for: '{query}'")
return f"data_result_for_{query}" # Simulated data result
class LoggingDecorator:
"""Wraps any service and adds logging around every call."""
def __init__(self, service):
self._service = service # The wrapped service object
def fetch(self, query: str) -> str:
print(f"[LOG] fetch() called with query='{query}'")
result = self._service.fetch(query) # Delegate to the real service
print(f"[LOG] fetch() returned '{result}'")
return result
class CachingDecorator:
"""Wraps any service and adds in-memory caching."""
def __init__(self, service):
self._service = service
self._cache = {} # Dictionary maps query → cached result
def fetch(self, query: str) -> str:
if query in self._cache:
print(f"[CACHE] HIT for '{query}'")
return self._cache[query] # Return cached result without calling service
print(f"[CACHE] MISS for '{query}'")
result = self._service.fetch(query) # Call service on cache miss
self._cache[query] = result # Store result in cache
return result
Stacking decorators at runtime:
service = DataService()
# Wrap with caching, then logging — order matters!
service = CachingDecorator(service)
service = LoggingDecorator(service)
service.fetch("user:42") # First call
service.fetch("user:42") # Second call — should hit cache
Output:
[LOG] fetch() called with query='user:42'
[CACHE] MISS for 'user:42'
Fetching data for: 'user:42'
[LOG] fetch() returned 'data_result_for_user:42'
[LOG] fetch() called with query='user:42'
[CACHE] HIT for 'user:42'
[LOG] fetch() returned 'data_result_for_user:42'
The problem: When an object's state changes, multiple other objects need to react — but the object shouldn't be tightly coupled to each of them.
The solution: The subject maintains a list of observers and notifies them all automatically when its state changes.
from abc import ABC, abstractmethod
class Observer(ABC):
"""Every observer must implement update()."""
@abstractmethod
def update(self, event: str, data: dict):
pass
class StockMarket:
"""The subject — notifies observers of price changes."""
def __init__(self):
self._observers = [] # List of registered observer objects
self._prices = {} # Current stock prices
def subscribe(self, observer: Observer):
self._observers.append(observer) # Register a new observer
def unsubscribe(self, observer: Observer):
self._observers.remove(observer) # Remove an observer
def _notify(self, event: str, data: dict):
# Broadcast event to every registered observer
for observer in self._observers:
observer.update(event, data)
def update_price(self, ticker: str, price: float):
old_price = self._prices.get(ticker, 0)
self._prices[ticker] = price
# Notify all observers of the price change
self._notify("price_update", {
"ticker": ticker,
"old_price": old_price,
"new_price": price,
"change_pct": ((price - old_price) / old_price * 100) if old_price else 0
})
class PriceAlertObserver(Observer):
def __init__(self, name: str, threshold_pct: float):
self.name = name # Alert name
self.threshold_pct = threshold_pct # Minimum change % to trigger alert
def update(self, event: str, data: dict):
if event == "price_update":
change = abs(data["change_pct"])
if change >= self.threshold_pct:
direction = "UP" if data["new_price"] > data["old_price"] else "DOWN"
print(f"[ALERT:{self.name}] {data['ticker']} moved {direction} "
f"{change:.1f}% to ${data['new_price']:.2f}")
class PortfolioObserver(Observer):
def __init__(self, holdings: dict):
self.holdings = holdings # Dict: ticker → number of shares
def update(self, event: str, data: dict):
if event == "price_update" and data["ticker"] in self.holdings:
shares = self.holdings[data["ticker"]]
value = shares * data["new_price"]
print(f"[PORTFOLIO] {data['ticker']} × {shares} shares = ${value:.2f}")
Live demo:
market = StockMarket()
market.subscribe(PriceAlertObserver("BigMover", threshold_pct=5.0))
market.subscribe(PortfolioObserver({"AAPL": 10, "GOOG": 5}))
market.update_price("AAPL", 150.00) # Establish baseline
market.update_price("AAPL", 162.00) # 8% increase
Output:
[PORTFOLIO] AAPL × 10 shares = $1500.00
[ALERT:BigMover] AAPL moved UP 8.0% to $162.00
[PORTFOLIO] AAPL × 10 shares = $1620.00
The problem: A class needs to do something (sort, compress, validate) but the exact algorithm should be swappable at runtime.
The solution: Encapsulate each algorithm in its own class behind a common interface. Inject the desired strategy.
from abc import ABC, abstractmethod
class SortStrategy(ABC):
@abstractmethod
def sort(self, data: list) -> list:
pass
class BubbleSortStrategy(SortStrategy):
def sort(self, data: list) -> list:
arr = data.copy() # Don't modify the original
n = len(arr)
for i in range(n):
for j in range(0, n - i - 1):
if arr[j] > arr[j + 1]:
arr[j], arr[j + 1] = arr[j + 1], arr[j] # Swap
return arr
class PythonSortStrategy(SortStrategy):
def sort(self, data: list) -> list:
return sorted(data) # Delegate to Python's built-in Timsort
class ReverseSortStrategy(SortStrategy):
def sort(self, data: list) -> list:
return sorted(data, reverse=True) # Sort descending
class DataProcessor:
def __init__(self, strategy: SortStrategy):
self._strategy = strategy # Injected strategy
def set_strategy(self, strategy: SortStrategy):
self._strategy = strategy # Swap strategy at runtime
def process(self, data: list) -> list:
print(f"Sorting with {type(self._strategy).__name__}")
return self._strategy.sort(data)
numbers = [5, 2, 8, 1, 9, 3]
processor = DataProcessor(PythonSortStrategy())
print(processor.process(numbers)) # [1, 2, 3, 5, 8, 9]
processor.set_strategy(ReverseSortStrategy())
print(processor.process(numbers)) # [9, 8, 5, 3, 2, 1]
| Pattern | Type | Problem Solved | Real Use Case |
|---|---|---|---|
| Adapter | Structural | Incompatible interface | Third-party API integration |
| Decorator | Structural | Add behavior without subclassing | Logging, caching, rate limiting |
| Observer | Behavioral | Notify many on state change | Event systems, stock alerts, UI events |
| Strategy | Behavioral | Swap algorithm at runtime | Sorting, compression, payment routing |
The electrical adapter didn't change your device. It didn't change the socket. It just bridged the gap. That's what structural patterns do.
Get this course's notes on Telegram!
Free cheat sheets, summaries & practice exercises