Design Patterns in Python - Idiomatic Implementations for Production Code

Reading time: ~35 minutes | Level: Intermediate → Engineering

Before reading further, consider why these two implementations of the same pattern are both "correct" in Python:

# Version A: classic OOP Strategy pattern
class SortStrategy:
    def sort(self, data: list) -> list:
        raise NotImplementedError

class BubbleSort(SortStrategy):
    def sort(self, data: list) -> list:
        d = list(data)
        for i in range(len(d)):
            for j in range(len(d) - i - 1):
                if d[j] > d[j+1]:
                    d[j], d[j+1] = d[j+1], d[j]
        return d

class Sorter:
    def __init__(self, strategy: SortStrategy):
        self.strategy = strategy

    def sort(self, data: list) -> list:
        return self.strategy.sort(data)

# Version B: Python-idiomatic - functions as strategies
from typing import Callable

def bubble_sort(data: list) -> list:
    d = list(data)
    for i in range(len(d)):
        for j in range(len(d) - i - 1):
            if d[j] > d[j+1]:
                d[j], d[j+1] = d[j+1], d[j]
    return d

class Sorter:
    def __init__(self, strategy: Callable[[list], list]):
        self.strategy = strategy

    def sort(self, data: list) -> list:
        return self.strategy(data)

Both are correct implementations of the Strategy pattern. Version B is more Pythonic. Understanding why - and when Version A is actually better - is what this lesson teaches.

Design patterns are solutions to recurring design problems. In Python, dynamic typing, first-class functions, decorators, and metaclasses mean the canonical Java/C++ implementations often look different. This lesson covers the patterns as ideas, then shows their idiomatic Python form.

What You Will Learn

• Why GoF patterns look different in Python than in Java/C++
• Singleton: the classic implementation and why module-level is usually better
• Factory and Abstract Factory: Python idioms using classmethods and Protocols
• Strategy: via callable or Protocol - when each is appropriate
• Observer: building an event system from scratch
• Decorator pattern vs Python decorator syntax - they are not the same thing
• Registry pattern: how Flask and Django use it for routing and app registry
• Builder pattern: fluent interfaces for complex object construction

Prerequisites

• Lessons 01–11 of this module
• Understanding of typing.Protocol, ABCs, decorators, and @dataclass
• Basic familiarity with Flask or Django (helpful but not required)

Part 1 - Why Python Patterns Look Different

The Core Difference: Dynamic Typing and First-Class Functions

In Java and C++, patterns often solve problems caused by the type system's rigidity. Python's type system is more flexible, which means:

1. Strategy does not need a class hierarchy - a callable works
2. Singleton does not need a static factory - module-level state works
3. Iterator is built into the language via __iter__/__next__
4. Template Method can be replaced by a higher-order function
5. Null Object is often just None with duck typing

Some patterns remain valuable in Python because they solve design problems unrelated to the type system: Observer, Factory, Registry, Builder. Others are patterns for specific constraints that Python does not have.

# Java needs a full class hierarchy for Strategy.
# Python: a function IS the strategy.

import sorted   # Python's sorted() is Strategy pattern built in:
# key= parameter accepts any callable - the sort strategy

data = ["banana", "apple", "cherry"]
print(sorted(data))                      # ['apple', 'banana', 'cherry']
print(sorted(data, key=len))             # ['apple', 'banana', 'cherry'] (by length)
print(sorted(data, key=lambda s: s[-1])) # sort by last char - strategy as lambda

sorted's key parameter is the Strategy pattern. No class hierarchy required.

note

Python's first-class functions make the Strategy pattern often simpler than the GoF version. When a strategy is a pure function with no state, pass it directly as a callable - no class, no interface, no inheritance needed. Use a Protocol or ABC only when the strategy carries state, configuration, or multiple related methods. The deciding question: does the strategy need __init__? If yes, use a class. If no, use a function.

Part 2 - Singleton Pattern

GoF Intent

Ensure a class has only one instance, and provide a global point of access to it.

The Classic Implementation

class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, value: int = 0):
        # __init__ runs every time, even when __new__ returns existing instance
        # Guard against re-initialisation:
        if not hasattr(self, "_initialised"):
            self.value = value
            self._initialised = True


a = Singleton(42)
b = Singleton(99)  # Does NOT re-initialise

print(a is b)      # True - same object
print(a.value)     # 42 - original value preserved
print(b.value)     # 42 - same object

Thread-Safe Singleton

import threading

class ThreadSafeSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                # Double-checked locking
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

The Pythonic Alternative: Module-Level State

In Python, a module is itself a singleton. Any module-level object is initialised once and shared across all imports.

# config.py - this IS a singleton
import os

class _Config:
    def __init__(self):
        self.debug = os.getenv("DEBUG", "false").lower() == "true"
        self.db_url = os.getenv("DATABASE_URL", "sqlite:///dev.db")
        self.secret_key = os.getenv("SECRET_KEY", "dev-secret")

# Instantiate once at module level
config = _Config()

# Any file that does 'from config import config' gets the SAME object

# app.py
from config import config

print(config.debug)     # False (or from env)
print(config.db_url)    # "sqlite:///dev.db"

# service.py
from config import config

# Same object - Python's module system guarantees this

Use module-level singleton for: application configuration, database connection pools, logger instances, feature flag clients.

warning

Singleton in Python: module-level globals ARE singletons; a class-based Singleton is usually over-engineering. Python's module import system guarantees a module is initialised once and its top-level objects are shared across all importers. Writing a custom __new__-based Singleton adds boilerplate and complexity with no benefit over config = _Config() at module level. Use the class-based Singleton only when you specifically need singleton behaviour tied to a class that is also used for subclassing or testing.

Singleton via Metaclass

For frameworks that need a more formal singleton mechanism:

class SingletonMeta(type):
    _instances: dict = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class DatabasePool(metaclass=SingletonMeta):
    def __init__(self, url: str = "postgresql://localhost/myapp"):
        self.url = url
        self._connections: list = []
        print(f"Pool created for {url}")

pool1 = DatabasePool()
pool2 = DatabasePool()
print(pool1 is pool2)   # True - "Pool created" printed only once

When NOT to Use Singleton

Singletons are global state. Global state makes code hard to test (tests share state) and hard to reason about (any code path can modify the singleton). Prefer dependency injection of a single shared instance rather than a true Singleton.

Part 3 - Factory and Abstract Factory

Factory Method: GoF Intent

Define an interface for creating an object, but let subclasses decide which class to instantiate.

Python Idiom: @classmethod as Factory

from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
import json

@dataclass
class DatabaseConfig:
    driver: str
    host: str
    port: int
    name: str

    @classmethod
    def from_env(cls) -> DatabaseConfig:
        import os
        return cls(
            driver=os.getenv("DB_DRIVER", "postgresql"),
            host=os.getenv("DB_HOST", "localhost"),
            port=int(os.getenv("DB_PORT", "5432")),
            name=os.getenv("DB_NAME", "myapp"),
        )

    @classmethod
    def from_url(cls, url: str) -> DatabaseConfig:
        """Parse postgresql://user:pass@host:port/name"""
        from urllib.parse import urlparse
        parsed = urlparse(url)
        return cls(
            driver=parsed.scheme,
            host=parsed.hostname or "localhost",
            port=parsed.port or 5432,
            name=parsed.path.lstrip("/"),
        )

    @classmethod
    def from_dict(cls, data: dict) -> DatabaseConfig:
        return cls(
            driver=data["driver"],
            host=data["host"],
            port=data["port"],
            name=data["name"],
        )

    @classmethod
    def sqlite(cls, path: str = ":memory:") -> DatabaseConfig:
        return cls(driver="sqlite", host="", port=0, name=path)


# Multiple factory methods - different creation paths
cfg1 = DatabaseConfig.from_url("postgresql://localhost:5432/mydb")
cfg2 = DatabaseConfig.from_env()
cfg3 = DatabaseConfig.sqlite(":memory:")

Factory Function: Choosing the Right Subclass

from typing import Protocol

class Cache(Protocol):
    def get(self, key: str) -> object | None: ...
    def set(self, key: str, value: object) -> None: ...

class InMemoryCache:
    def __init__(self):
        self._store: dict = {}
    def get(self, key: str) -> object | None:
        return self._store.get(key)
    def set(self, key: str, value: object) -> None:
        self._store[key] = value

class RedisCache:
    def __init__(self, host: str, port: int):
        self._host = host
        self._port = port
    def get(self, key: str) -> object | None: ...
    def set(self, key: str, value: object) -> None: ...

class NullCache:
    """No-op cache for testing."""
    def get(self, key: str) -> object | None:
        return None
    def set(self, key: str, value: object) -> None:
        pass

def create_cache(cache_type: str, **kwargs) -> Cache:
    """Factory function - returns the appropriate Cache implementation."""
    match cache_type:
        case "memory":
            return InMemoryCache()
        case "redis":
            return RedisCache(
                host=kwargs.get("host", "localhost"),
                port=kwargs.get("port", 6379),
            )
        case "null":
            return NullCache()
        case _:
            raise ValueError(f"Unknown cache type: {cache_type!r}")

cache = create_cache("redis", host="redis.example.com", port=6379)

The Factory pattern's decision flow looks like this:

Abstract Factory: Families of Related Objects

from abc import ABC, abstractmethod

class Button(ABC):
    @abstractmethod
    def render(self) -> str: ...

class TextInput(ABC):
    @abstractmethod
    def render(self) -> str: ...

class UIFactory(ABC):
    """Abstract Factory: creates a family of related UI components."""
    @abstractmethod
    def create_button(self, label: str) -> Button: ...
    @abstractmethod
    def create_text_input(self, placeholder: str) -> TextInput: ...

# Concrete family: Bootstrap
class BootstrapButton(Button):
    def __init__(self, label: str):
        self._label = label
    def render(self) -> str:
        return f'<button class="btn btn-primary">{self._label}</button>'

class BootstrapTextInput(TextInput):
    def __init__(self, placeholder: str):
        self._placeholder = placeholder
    def render(self) -> str:
        return f'<input class="form-control" placeholder="{self._placeholder}">'

class BootstrapFactory(UIFactory):
    def create_button(self, label: str) -> Button:
        return BootstrapButton(label)
    def create_text_input(self, placeholder: str) -> TextInput:
        return BootstrapTextInput(placeholder)

# Concrete family: Tailwind
class TailwindButton(Button):
    def __init__(self, label: str):
        self._label = label
    def render(self) -> str:
        return f'<button class="px-4 py-2 bg-blue-500 text-white">{self._label}</button>'

class TailwindTextInput(TextInput):
    def __init__(self, placeholder: str):
        self._placeholder = placeholder
    def render(self) -> str:
        return f'<input class="border rounded px-3 py-2" placeholder="{self._placeholder}">'

class TailwindFactory(UIFactory):
    def create_button(self, label: str) -> Button:
        return TailwindButton(label)
    def create_text_input(self, placeholder: str) -> TextInput:
        return TailwindTextInput(placeholder)

# High-level code uses only the abstract factory
def render_login_form(factory: UIFactory) -> str:
    email_input = factory.create_text_input("Email address")
    password_input = factory.create_text_input("Password")
    submit_button = factory.create_button("Sign In")
    return f"""
<form>
  {email_input.render()}
  {password_input.render()}
  {submit_button.render()}
</form>"""

print(render_login_form(BootstrapFactory()))
print(render_login_form(TailwindFactory()))
# Same structure, different styling - no conditional in render_login_form

Part 4 - Strategy Pattern

GoF Intent

Define a family of algorithms, encapsulate each one, and make them interchangeable.

Python with Callable (Preferred for Simple Strategies)

from typing import Callable, TypeVar

T = TypeVar("T")

class Sorter:
    def __init__(self, strategy: Callable[[list], list] = sorted):
        self._strategy = strategy

    def sort(self, data: list) -> list:
        return self._strategy(data)

# Strategies are just functions - no class hierarchy needed
def reverse_sort(data: list) -> list:
    return sorted(data, reverse=True)

def length_sort(data: list) -> list:
    return sorted(data, key=len)

sorter = Sorter(strategy=reverse_sort)
print(sorter.sort(["banana", "apple", "cherry"]))
# ['cherry', 'banana', 'apple']

sorter = Sorter(strategy=length_sort)
print(sorter.sort(["banana", "apple", "cherry"]))
# ['apple', 'banana', 'cherry']

tip

The Strategy pattern replaces if/elif chains with polymorphism, making the system easier to extend without touching existing code. Before Strategy: if type == "bulk": ... elif type == "loyalty": ... - adding "seasonal" pricing means modifying this block. After Strategy: add a SeasonalPricing class and inject it. Zero changes to OrderCalculator. This is OCP in action, implemented via Strategy.

Python with Protocol (When Strategies Have State)

When the strategy needs state (configuration, connections, prior results), use a Protocol or ABC:

from typing import Protocol

class PricingStrategy(Protocol):
    def calculate(self, base_price: float, quantity: int) -> float: ...

class StandardPricing:
    def calculate(self, base_price: float, quantity: int) -> float:
        return base_price * quantity

class BulkDiscountPricing:
    def __init__(self, discount_threshold: int, discount_rate: float):
        self._threshold = discount_threshold
        self._rate = discount_rate

    def calculate(self, base_price: float, quantity: int) -> float:
        total = base_price * quantity
        if quantity >= self._threshold:
            total *= (1 - self._rate)
        return total

class LoyaltyPricing:
    def __init__(self, customer_points: int):
        self._points = customer_points

    def calculate(self, base_price: float, quantity: int) -> float:
        discount = min(self._points * 0.001, 0.20)  # max 20% from points
        return base_price * quantity * (1 - discount)

class OrderCalculator:
    def __init__(self, pricing: PricingStrategy):
        self._pricing = pricing

    def total(self, base_price: float, quantity: int) -> float:
        return self._pricing.calculate(base_price, quantity)

calc = OrderCalculator(BulkDiscountPricing(discount_threshold=10, discount_rate=0.15))
print(calc.total(base_price=50.0, quantity=12))   # 510.0 (15% off)

calc = OrderCalculator(LoyaltyPricing(customer_points=500))
print(calc.total(base_price=50.0, quantity=5))    # 237.5 (5% off from points)

Rule: use a callable when the strategy is a pure function (no state, no configuration). Use a Protocol or ABC when the strategy carries state or configuration.

Part 5 - Observer Pattern

GoF Intent

Define a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically.

Event System Implementation

from typing import Callable, Any
from collections import defaultdict

class EventBus:
    """Central event dispatcher - Observer pattern."""

    def __init__(self):
        self._listeners: dict[str, list[Callable]] = defaultdict(list)

    def subscribe(self, event_type: str, handler: Callable[..., Any]) -> None:
        self._listeners[event_type].append(handler)

    def unsubscribe(self, event_type: str, handler: Callable[..., Any]) -> None:
        self._listeners[event_type].remove(handler)

    def publish(self, event_type: str, **payload) -> None:
        for handler in self._listeners[event_type]:
            handler(**payload)


# Observers - just functions, no class hierarchy needed
def send_welcome_email(username: str, email: str, **kwargs) -> None:
    print(f"EMAIL → {email}: Welcome, {username}!")

def create_default_settings(username: str, **kwargs) -> None:
    print(f"Settings: Creating defaults for {username}")

def log_registration(username: str, email: str, **kwargs) -> None:
    print(f"AUDIT: New user registered: {username} ({email})")

# Wire up
bus = EventBus()
bus.subscribe("user.registered", send_welcome_email)
bus.subscribe("user.registered", create_default_settings)
bus.subscribe("user.registered", log_registration)

# Subject publishes - observers react
bus.publish("user.registered", username="alice", email="[email protected]")
# EMAIL → [email protected]: Welcome, alice!
# Settings: Creating defaults for alice
# AUDIT: New user registered: alice ([email protected])

Typed Observer with Dataclasses

from dataclasses import dataclass
from typing import Callable, Generic, TypeVar

E = TypeVar("E")

@dataclass
class UserRegisteredEvent:
    username: str
    email: str
    user_id: int

@dataclass
class OrderPlacedEvent:
    order_id: str
    user_id: int
    total: float

class TypedEventBus(Generic[E]):
    def __init__(self):
        self._handlers: list[Callable[[E], None]] = []

    def subscribe(self, handler: Callable[[E], None]) -> None:
        self._handlers.append(handler)

    def publish(self, event: E) -> None:
        for handler in self._handlers:
            handler(event)

user_bus: TypedEventBus[UserRegisteredEvent] = TypedEventBus()

def on_user_registered(event: UserRegisteredEvent) -> None:
    print(f"New user: {event.username} (id={event.user_id})")

user_bus.subscribe(on_user_registered)
user_bus.publish(UserRegisteredEvent(username="alice", email="[email protected]", user_id=1))

Framework Usage: Django Signals

Django's signal system is the Observer pattern:

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User

@receiver(post_save, sender=User)
def on_user_created(sender, instance: User, created: bool, **kwargs) -> None:
    """Observed: fires after any User is saved."""
    if created:
        print(f"New user registered: {instance.username}")
        # send welcome email, create profile, etc.

Part 6 - Decorator Pattern vs Python Decorator Syntax

The Confusion

The GoF Decorator pattern and Python's @decorator syntax are NOT the same thing. They solve similar problems but are implemented differently.

GoF Decorator pattern: wraps an object to add behaviour without modifying the original class.

Python @decorator syntax: syntactic sugar for applying a higher-order function to a function or class.

GoF Decorator Pattern in Python

from typing import Protocol

class TextFormatter(Protocol):
    def format(self, text: str) -> str: ...

class PlainText:
    def format(self, text: str) -> str:
        return text

class BoldDecorator:
    def __init__(self, component: TextFormatter):
        self._component = component

    def format(self, text: str) -> str:
        return f"<b>{self._component.format(text)}</b>"

class ItalicDecorator:
    def __init__(self, component: TextFormatter):
        self._component = component

    def format(self, text: str) -> str:
        return f"<i>{self._component.format(text)}</i>"

class UppercaseDecorator:
    def __init__(self, component: TextFormatter):
        self._component = component

    def format(self, text: str) -> str:
        return self._component.format(text).upper()

# Compose decorators - order matters
text = PlainText()
bold = BoldDecorator(text)
italic_bold = ItalicDecorator(bold)

print(text.format("hello"))         # hello
print(bold.format("hello"))         # <b>hello</b>
print(italic_bold.format("hello"))  # <i><b>hello</b></i>

# Unlimited composition
from_scratch = UppercaseDecorator(ItalicDecorator(BoldDecorator(PlainText())))
print(from_scratch.format("hello")) # <I><B>HELLO</B></I>

Python Function Decorators as the Decorator Pattern

Python's functools.wraps-based decorators are a more common implementation:

import functools
import time
from typing import Callable, TypeVar, Any

F = TypeVar("F", bound=Callable[..., Any])

def timeit(func: F) -> F:
    """Decorator: wraps a function to measure execution time."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper  # type: ignore[return-value]

def retry(max_attempts: int = 3, exceptions: tuple = (Exception,)):
    """Parameterised decorator: adds retry logic to any function."""
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_error = e
                    print(f"Attempt {attempt}/{max_attempts} failed: {e}")
            raise last_error
        return wrapper  # type: ignore[return-value]
    return decorator

def require_auth(func: F) -> F:
    """Decorator: checks authentication before calling function."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # In a real app: check request.user.is_authenticated
        print("Auth check passed")
        return func(*args, **kwargs)
    return wrapper  # type: ignore[return-value]


@timeit
@retry(max_attempts=3, exceptions=(ConnectionError,))
@require_auth
def fetch_user_data(user_id: int) -> dict:
    return {"id": user_id, "name": "Alice"}

result = fetch_user_data(1)
# Auth check passed
# fetch_user_data took 0.0001s

The stacking of decorators is the GoF Decorator pattern: each decorator adds behaviour to the wrapped object without modifying it.

danger

God Object anti-pattern: a class that knows too much and does too much is the opposite of good pattern use. A God Object accumulates methods, state, and responsibilities that belong in separate classes - often the result of repeatedly extending a single class instead of applying Strategy, Observer, or Decorator. Signs of a God Object: the class has more than ~10 methods, it imports from 5+ modules, it is described with multiple "and"s. Break it up: extract strategies, observers, and decorators into focused components.

Part 7 - Registry Pattern

GoF Intent (Component/Registry)

Maintain a registry of objects or classes, allowing lookup by name or key. This enables plugins, extensible systems, and decoupled component registration.

Basic Registry

from typing import Type, TypeVar

T = TypeVar("T")

class Registry:
    """Central registry - maps string keys to classes."""

    _registry: dict[str, type] = {}

    @classmethod
    def register(cls, name: str):
        """Decorator: register a class under a name."""
        def decorator(klass: type) -> type:
            cls._registry[name] = klass
            return klass
        return decorator

    @classmethod
    def create(cls, name: str, *args, **kwargs):
        """Factory: create an instance of the registered class."""
        if name not in cls._registry:
            raise KeyError(f"No class registered for {name!r}. "
                           f"Available: {list(cls._registry.keys())}")
        return cls._registry[name](*args, **kwargs)

    @classmethod
    def available(cls) -> list[str]:
        return list(cls._registry.keys())


class StorageRegistry(Registry):
    _registry: dict[str, type] = {}   # separate registry per subclass

@StorageRegistry.register("local")
class LocalStorage:
    def __init__(self, path: str):
        self.path = path
    def read(self, key: str) -> bytes:
        with open(f"{self.path}/{key}", "rb") as f:
            return f.read()

@StorageRegistry.register("s3")
class S3Storage:
    def __init__(self, bucket: str):
        self.bucket = bucket
    def read(self, key: str) -> bytes:
        print(f"S3: reading {self.bucket}/{key}")
        return b"data"

@StorageRegistry.register("memory")
class MemoryStorage:
    def __init__(self):
        self._store: dict = {}
    def read(self, key: str) -> bytes:
        return self._store.get(key, b"")

# Create by name - the type is determined at runtime
storage = StorageRegistry.create("memory")
print(type(storage).__name__)   # MemoryStorage

print(StorageRegistry.available())   # ['local', 's3', 'memory']

Flask Route Registry

Flask's @app.route decorator is the Registry pattern:

from flask import Flask, request, jsonify

app = Flask(__name__)

# @app.route registers the function in Flask's URL map (a registry)
@app.route("/users", methods=["GET"])
def list_users():
    return jsonify([{"id": 1, "name": "Alice"}])

@app.route("/users/<int:user_id>", methods=["GET"])
def get_user(user_id: int):
    return jsonify({"id": user_id, "name": "Alice"})

@app.route("/users", methods=["POST"])
def create_user():
    data = request.json
    return jsonify({"id": 2, **data}), 201

# Flask's internal registry: {"/users GET": list_users, "/users POST": create_user, ...}

Django App Registry

Django's AppConfig and app_registry system is also the Registry pattern:

# myapp/apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = "myapp"
    label = "myapp"

    def ready(self):
        # This hook fires when the app is loaded into Django's registry
        import myapp.signals  # connect signal handlers at registration time

# Django's apps.get_model("myapp", "User") - Registry lookup by app label + model name

Plugin System Using Registry

from typing import Protocol, Type

class Plugin(Protocol):
    def execute(self, context: dict) -> dict: ...

class PluginRegistry:
    _plugins: dict[str, type] = {}

    @classmethod
    def plugin(cls, name: str):
        """Decorator to register a plugin class."""
        def decorator(klass: type) -> type:
            cls._plugins[name] = klass
            print(f"Plugin registered: {name}")
            return klass
        return decorator

    @classmethod
    def run(cls, name: str, context: dict) -> dict:
        if name not in cls._plugins:
            raise KeyError(f"Plugin not found: {name!r}")
        return cls._plugins[name]().execute(context)

@PluginRegistry.plugin("uppercase")
class UppercasePlugin:
    def execute(self, context: dict) -> dict:
        return {k: v.upper() if isinstance(v, str) else v for k, v in context.items()}

@PluginRegistry.plugin("reverse")
class ReversePlugin:
    def execute(self, context: dict) -> dict:
        return {k: v[::-1] if isinstance(v, str) else v for k, v in context.items()}

result = PluginRegistry.run("uppercase", {"name": "alice", "city": "london"})
print(result)   # {'name': 'ALICE', 'city': 'LONDON'}

Part 8 - Builder Pattern

GoF Intent

Separate the construction of a complex object from its representation, so the same construction process can create different representations.

Fluent Builder

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class QueryBuilder:
    """Fluent builder for SQL SELECT queries."""

    _table: str = ""
    _columns: list[str] = field(default_factory=list)
    _conditions: list[str] = field(default_factory=list)
    _order_by: Optional[str] = None
    _limit: Optional[int] = None
    _offset: Optional[int] = None
    _joins: list[str] = field(default_factory=list)

    def from_table(self, table: str) -> QueryBuilder:
        self._table = table
        return self   # return self for chaining

    def select(self, *columns: str) -> QueryBuilder:
        self._columns.extend(columns)
        return self

    def where(self, condition: str) -> QueryBuilder:
        self._conditions.append(condition)
        return self

    def join(self, table: str, on: str) -> QueryBuilder:
        self._joins.append(f"JOIN {table} ON {on}")
        return self

    def order_by(self, column: str, direction: str = "ASC") -> QueryBuilder:
        self._order_by = f"{column} {direction}"
        return self

    def limit(self, n: int) -> QueryBuilder:
        self._limit = n
        return self

    def offset(self, n: int) -> QueryBuilder:
        self._offset = n
        return self

    def build(self) -> str:
        if not self._table:
            raise ValueError("Table must be specified via .from_table()")
        cols = ", ".join(self._columns) if self._columns else "*"
        sql = f"SELECT {cols} FROM {self._table}"
        for join in self._joins:
            sql += f"\n{join}"
        if self._conditions:
            sql += "\nWHERE " + " AND ".join(self._conditions)
        if self._order_by:
            sql += f"\nORDER BY {self._order_by}"
        if self._limit is not None:
            sql += f"\nLIMIT {self._limit}"
        if self._offset is not None:
            sql += f"\nOFFSET {self._offset}"
        return sql


# Fluent interface: each method returns self for chaining
query = (
    QueryBuilder()
    .from_table("users")
    .select("id", "username", "email")
    .join("orders", on="orders.user_id = users.id")
    .where("users.active = TRUE")
    .where("users.created_at > '2024-01-01'")
    .order_by("username")
    .limit(20)
    .offset(40)
    .build()
)
print(query)
# SELECT id, username, email FROM users
# JOIN orders ON orders.user_id = users.id
# WHERE users.active = TRUE AND users.created_at > '2024-01-01'
# ORDER BY username ASC
# LIMIT 20
# OFFSET 40

Builder for Configuration Objects

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ServerConfig:
    host: str
    port: int
    workers: int
    debug: bool
    tls_enabled: bool
    tls_cert: Optional[str]
    tls_key: Optional[str]
    max_connections: int
    keepalive_timeout: int
    access_log: bool
    error_log_path: str

class ServerConfigBuilder:
    def __init__(self):
        self._host = "0.0.0.0"
        self._port = 8000
        self._workers = 4
        self._debug = False
        self._tls_enabled = False
        self._tls_cert = None
        self._tls_key = None
        self._max_connections = 1000
        self._keepalive_timeout = 65
        self._access_log = True
        self._error_log_path = "/var/log/app/error.log"

    def host(self, host: str) -> ServerConfigBuilder:
        self._host = host
        return self

    def port(self, port: int) -> ServerConfigBuilder:
        self._port = port
        return self

    def workers(self, n: int) -> ServerConfigBuilder:
        self._workers = n
        return self

    def debug(self) -> ServerConfigBuilder:
        self._debug = True
        return self

    def with_tls(self, cert: str, key: str) -> ServerConfigBuilder:
        self._tls_enabled = True
        self._tls_cert = cert
        self._tls_key = key
        return self

    def max_connections(self, n: int) -> ServerConfigBuilder:
        self._max_connections = n
        return self

    def build(self) -> ServerConfig:
        if self._tls_enabled and (not self._tls_cert or not self._tls_key):
            raise ValueError("TLS requires both cert and key paths")
        return ServerConfig(
            host=self._host,
            port=self._port,
            workers=self._workers,
            debug=self._debug,
            tls_enabled=self._tls_enabled,
            tls_cert=self._tls_cert,
            tls_key=self._tls_key,
            max_connections=self._max_connections,
            keepalive_timeout=self._keepalive_timeout,
            access_log=self._access_log,
            error_log_path=self._error_log_path,
        )


# Production config
prod_config = (
    ServerConfigBuilder()
    .host("0.0.0.0")
    .port(443)
    .workers(8)
    .with_tls("/etc/ssl/cert.pem", "/etc/ssl/key.pem")
    .max_connections(5000)
    .build()
)

# Development config
dev_config = (
    ServerConfigBuilder()
    .port(8000)
    .debug()
    .build()
)

Part 9 - Choosing the Right Pattern

A quick decision guide for the patterns in this lesson:

Problem	Pattern	Python-first approach
Ensure one instance exists	Singleton	Module-level object
Create objects without knowing the exact class	Factory Method	@classmethod or factory function
Create families of related objects	Abstract Factory	ABC + concrete factories
Swap algorithms at runtime	Strategy	Callable or Protocol
React to state changes without tight coupling	Observer	EventBus or Django signals
Add behaviour to an object dynamically	Decorator	functools.wraps-based decorator
Map names to classes/functions	Registry	dict + decorator @registry.register
Construct complex objects step by step	Builder	Fluent methods returning self

Common Mistakes

Mistake 1 - Singleton When You Need Dependency Injection

# Singleton: global, hard to test, hard to swap
class Database:
    _instance = None
    def __new__(cls): ...

db = Database()   # global - every test shares this

# Better: inject a single shared instance
db = Database(url="postgresql://localhost/myapp")
service = UserService(db=db)   # injected - tests can inject a mock

Mistake 2 - Factory That Is Just a Constructor with Extra Steps

# Unnecessary factory - just use the constructor
class UserFactory:
    def create(self, username, email):
        return User(username=username, email=email)

# Factories add value when they: choose between implementations,
# read from external sources, or apply non-trivial construction logic

Mistake 3 - Observer Without Unsubscribe (Memory Leak)

class EventBus:
    def subscribe(self, event: str, handler): ...
    # Missing unsubscribe!

# If handlers hold references to long-lived objects
# and the bus holds references to handlers,
# objects cannot be garbage collected - memory leak

# Fix: provide unsubscribe() and use weakref for handlers when appropriate
import weakref

Mistake 4 - Builder Without Validation in build()

class QueryBuilder:
    def build(self) -> str:
        return f"SELECT * FROM {self._table}"   # no validation

# Calling build() on an incomplete builder should fail clearly
# Add validation: if not self._table: raise ValueError(...)

Mistake 5 - Using GoF Class Hierarchy Strategy When a Function Suffices

# Over-engineered
class SortStrategy(ABC):
    @abstractmethod
    def sort(self, data: list) -> list: ...

class AscendingSort(SortStrategy):
    def sort(self, data: list) -> list: return sorted(data)

# Pythonic
sorter = Sorter(strategy=sorted)            # standard library function
sorter = Sorter(strategy=lambda d: sorted(d, reverse=True))  # lambda

Engineering Checklist

Before moving to the next module, verify you can answer these without looking:

1. Why does the Strategy pattern look different in Python than in Java? What does Python's type system enable?
2. Why is module-level state usually preferable to a Singleton class in Python?
3. What is the difference between a Factory Method and an Abstract Factory?
4. When should a Strategy be a callable vs a Protocol? What is the deciding factor?
5. How does the Observer pattern differ from direct method calls, and what does it decouple?
6. What is the difference between the GoF Decorator pattern and Python's @decorator syntax?
7. What is the Registry pattern, and how does Flask's @app.route implement it?
8. When does the Builder pattern add value over a dataclass with a complex __post_init__?
9. What does functools.wraps do, and why is it important in function decorators?

Graded Practice Challenges

Level 1 - Predict the Output

Question 1

class Registry:
    _reg = {}

    @classmethod
    def register(cls, name):
        def decorator(klass):
            cls._reg[name] = klass
            return klass
        return decorator

    @classmethod
    def create(cls, name):
        return cls._reg[name]()

@Registry.register("alpha")
class Alpha:
    def greet(self): return "Hello from Alpha"

@Registry.register("beta")
class Beta:
    def greet(self): return "Hello from Beta"

obj = Registry.create("alpha")
print(obj.greet())
print(Registry.create("beta").greet())
print(list(Registry._reg.keys()))

Show Answer

Hello from Alpha
Hello from Beta
['alpha', 'beta']

@Registry.register("alpha") calls register("alpha") which returns a decorator. That decorator is applied to Alpha, storing it in _reg["alpha"] and returning the class unchanged. create("alpha") looks up the class and calls it with () to get an instance.

Question 2

from typing import Callable

class EventBus:
    def __init__(self):
        self._handlers = {}

    def on(self, event: str, handler: Callable):
        self._handlers.setdefault(event, []).append(handler)

    def emit(self, event: str, **data):
        for h in self._handlers.get(event, []):
            h(**data)

bus = EventBus()
results = []

bus.on("ping", lambda msg: results.append(f"A:{msg}"))
bus.on("ping", lambda msg: results.append(f"B:{msg}"))
bus.emit("ping", msg="hello")
bus.emit("pong", msg="ignored")
print(results)

Show Answer

['A:hello', 'B:hello']

Both handlers subscribed to "ping" are called with msg="hello". The "pong" event has no handlers, so self._handlers.get("pong", []) returns an empty list and nothing runs.

Question 3

import functools

def tag(name):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            return f"<{name}>{result}</{name}>"
        return wrapper
    return decorator

@tag("b")
@tag("i")
def greet(name):
    return f"Hello, {name}"

print(greet("World"))

Show Answer

<b><i>Hello, World</i></b>

Decorators are applied bottom-up: greet is first wrapped by @tag("i"), then by @tag("b"). Calling greet("World") calls the b wrapper, which calls the i wrapper, which calls the original, returning "Hello, World". The i wrapper wraps it in <i>...</i>, then the b wrapper wraps that in <b>...</b>.

Question 4

class Builder:
    def __init__(self):
        self._parts = []

    def add(self, part: str) -> "Builder":
        self._parts.append(part)
        return self

    def build(self) -> str:
        return " | ".join(self._parts)

b = Builder()
result = b.add("A").add("B").add("C").build()
print(result)
print(len(b._parts))

Show Answer

A | B | C
3

Each add() mutates self._parts and returns self, enabling method chaining. After three add() calls, _parts contains three elements. build() joins them with " | ". The builder object b still holds all three parts after build().

Question 5

class Singleton:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

a = Singleton()
b = Singleton()
a.value = 42
print(b.value)
print(a is b)

Show Answer

42
True

a and b are the same object (a is b is True). Setting a.value = 42 sets an attribute on the single shared instance. Accessing b.value reads from the same object, so it returns 42.

Level 2 - Debug Challenge

This Observer implementation has two bugs that will cause problems in production. Find and fix both:

from collections import defaultdict
from typing import Callable

class EventBus:
    def __init__(self):
        self._listeners = defaultdict(list)

    def subscribe(self, event: str, handler: Callable) -> None:
        self._listeners[event].append(handler)

    def publish(self, event: str, **payload) -> None:
        for handler in self._listeners[event]:   # Bug 1
            handler(**payload)

    def unsubscribe(self, event: str, handler: Callable) -> None:
        self._listeners[event].remove(handler)   # Bug 2


bus = EventBus()

def handler_a(msg): print(f"A: {msg}")
def handler_b(msg): print(f"B: {msg}")

bus.subscribe("test", handler_a)
bus.subscribe("test", handler_b)

# This will cause problems:
for h in [handler_a, handler_b]:
    bus.unsubscribe("test", h)
    bus.publish("test", msg="hello")

Show Solution

Bug 1 - Iterating over the listener list while it could be modified during dispatch: If a handler calls bus.unsubscribe() during publish(), the list is modified mid-iteration, causing a RuntimeError or skipped handlers. Fix: iterate over a copy.

def publish(self, event: str, **payload) -> None:
    for handler in list(self._listeners[event]):   # iterate a copy
        handler(**payload)

Bug 2 - list.remove() raises ValueError if handler is not subscribed: bus.unsubscribe("test", handler_a) on a handler that was already removed raises ValueError: list.remove(x): x not in list. Fix: check before removing or use a try/except.

def unsubscribe(self, event: str, handler: Callable) -> None:
    try:
        self._listeners[event].remove(handler)
    except ValueError:
        pass  # handler was not subscribed - safe to ignore

Full corrected class:

from collections import defaultdict
from typing import Callable

class EventBus:
    def __init__(self):
        self._listeners: dict[str, list[Callable]] = defaultdict(list)

    def subscribe(self, event: str, handler: Callable) -> None:
        self._listeners[event].append(handler)

    def publish(self, event: str, **payload) -> None:
        for handler in list(self._listeners[event]):   # copy to allow unsubscribe during dispatch
            handler(**payload)

    def unsubscribe(self, event: str, handler: Callable) -> None:
        try:
            self._listeners[event].remove(handler)
        except ValueError:
            pass

Level 3 - Design Challenge

Design a notification pipeline system using at least three design patterns from this lesson. The system must:

1. Support multiple notification channels (Email, SMS, Slack) - use Strategy or Registry
2. Enrich notifications with metadata (timestamp, correlation ID) before sending - use Decorator
3. Allow new channel types to be registered at runtime by name - use Registry
4. Use an Observer-style event bus so the pipeline is triggered by events, not direct calls
5. Include a reference to which patterns are used and why in comments

Show Reference Solution

import functools
import uuid
from datetime import datetime
from typing import Callable, Protocol
from collections import defaultdict

# === Strategy: notification channel protocol ===

class NotificationChannel(Protocol):
    def send(self, recipient: str, message: str, metadata: dict) -> bool: ...


# === Registry: channels registered by name ===

class ChannelRegistry:
    _channels: dict[str, type] = {}

    @classmethod
    def register(cls, name: str):
        """Registry pattern: register channel class by name."""
        def decorator(klass: type) -> type:
            cls._channels[name] = klass
            return klass
        return decorator

    @classmethod
    def create(cls, name: str, **kwargs) -> NotificationChannel:
        if name not in cls._channels:
            raise KeyError(f"Channel not registered: {name!r}. Available: {list(cls._channels)}")
        return cls._channels[name](**kwargs)

    @classmethod
    def available(cls) -> list[str]:
        return list(cls._channels.keys())


@ChannelRegistry.register("email")
class EmailChannel:
    def send(self, recipient: str, message: str, metadata: dict) -> bool:
        print(f"EMAIL [{metadata.get('correlation_id', 'N/A')}] → {recipient}: {message}")
        return True

@ChannelRegistry.register("sms")
class SMSChannel:
    def send(self, recipient: str, message: str, metadata: dict) -> bool:
        print(f"SMS  [{metadata.get('correlation_id', 'N/A')}] → {recipient}: {message}")
        return True

@ChannelRegistry.register("slack")
class SlackChannel:
    def __init__(self, webhook: str = "https://hooks.slack.com/default"):
        self._webhook = webhook

    def send(self, recipient: str, message: str, metadata: dict) -> bool:
        print(f"SLACK [{self._webhook}] → {recipient}: {message} (ts={metadata.get('timestamp')})")
        return True


# === Decorator pattern: enrich notification with metadata ===

def with_correlation_id(channel: NotificationChannel) -> NotificationChannel:
    """Decorator: adds correlation_id to every notification."""
    class CorrelationDecorator:
        def send(self, recipient: str, message: str, metadata: dict) -> bool:
            enriched = {**metadata, "correlation_id": str(uuid.uuid4())[:8]}
            return channel.send(recipient, message, enriched)
    return CorrelationDecorator()

def with_timestamp(channel: NotificationChannel) -> NotificationChannel:
    """Decorator: adds ISO timestamp to every notification."""
    class TimestampDecorator:
        def send(self, recipient: str, message: str, metadata: dict) -> bool:
            enriched = {**metadata, "timestamp": datetime.utcnow().isoformat()}
            return channel.send(recipient, message, enriched)
    return TimestampDecorator()


# === Observer: event bus triggers the pipeline ===

class NotificationEventBus:
    """Observer pattern: publish notification events to subscribed pipelines."""

    def __init__(self):
        self._handlers: dict[str, list[Callable]] = defaultdict(list)

    def subscribe(self, event: str, handler: Callable) -> None:
        self._handlers[event].append(handler)

    def publish(self, event: str, **payload) -> None:
        for handler in list(self._handlers[event]):
            handler(**payload)


# === Notification pipeline: wires Strategy + Registry + Decorator + Observer ===

class NotificationPipeline:
    def __init__(self, bus: NotificationEventBus, channel_names: list[str]):
        # Registry: resolve channels by name
        raw_channels = [ChannelRegistry.create(name) for name in channel_names]
        # Decorator: enrich all channels with correlation_id and timestamp
        self._channels = [
            with_timestamp(with_correlation_id(ch))
            for ch in raw_channels
        ]
        # Observer: subscribe to notification events
        bus.subscribe("notify", self._on_notify)

    def _on_notify(self, recipient: str, message: str, **kwargs) -> None:
        for channel in self._channels:
            channel.send(recipient, message, metadata={})


# === Usage ===

bus = NotificationEventBus()

# Pipeline uses email + sms - registered by name (Registry)
pipeline = NotificationPipeline(bus, channel_names=["email", "sms"])

# Add Slack pipeline later - no changes to existing code (OCP via Registry)
slack_pipeline = NotificationPipeline(bus, channel_names=["slack"])

# Trigger notification via event (Observer - not direct call)
bus.publish("notify", recipient="[email protected]", message="Your order has shipped!")

# Register a new channel type at runtime
@ChannelRegistry.register("webhook")
class WebhookChannel:
    def send(self, recipient: str, message: str, metadata: dict) -> bool:
        print(f"WEBHOOK → {recipient}: {message}")
        return True

print(f"\nAvailable channels: {ChannelRegistry.available()}")
# Available channels: ['email', 'sms', 'slack', 'webhook']

Patterns used:

• Registry: ChannelRegistry maps channel names to classes; @register("email") registers at class definition time; new channels added with zero changes to NotificationPipeline
• Strategy: each NotificationChannel is a strategy for delivering a notification; swappable without changing the pipeline
• Decorator (GoF): with_correlation_id and with_timestamp wrap any channel to add metadata; they compose: with_timestamp(with_correlation_id(channel))
• Observer: NotificationEventBus decouples the event source from the pipeline; the pipeline subscribes to "notify" events rather than being called directly

Key Takeaways

• GoF patterns exist to solve design problems. In Python, many problems Java patterns address are solved by the language itself - first-class functions, modules as singletons, structural typing via Protocol.
• Singleton: prefer module-level objects over class-based Singletons. A class-based Singleton is usually over-engineering in Python.
• Factory: @classmethod factory methods are the Pythonic idiom. Use factory functions when you need to choose between implementations at runtime.
• Strategy: use a callable when the strategy is a pure function. Use a Protocol when it carries state or configuration. Strategy replaces if/elif chains - extending the system means adding a new class, not modifying existing code.
• Observer: decouples the event producer from consumers. Handlers are registered and called without the producer knowing about them. Always implement unsubscribe() and iterate over a copy of the handler list during dispatch.
• Decorator pattern (GoF) is object composition: wrapping one object in another to add behaviour. Python's @decorator syntax is function wrapping - they are related but not identical.
• Registry: a dictionary of name → class. Used by Flask (routes), Django (apps, models), and any plugin-based system. @registry.register("name") is the canonical Pythonic idiom.
• Builder: fluent methods that return self for chaining. build() must validate that the object is in a complete state before returning the final object.
• God Object anti-pattern: the opposite of good design. A class that accumulates everything is the result of not applying patterns. Break it up using Strategy, Observer, or Decorator as appropriate.
• Python's first-class functions make Strategy often simpler than the GoF version - pass a function instead of a Strategy class whenever the strategy has no state.

What's Next

This completes the Object-Oriented Programming module. You now have the full engineering-depth OOP toolkit: the object model, construction, dunder methods, properties and descriptors, inheritance, MRO, ABCs, dataclasses, SOLID principles, and design patterns.

The next module is Functional Programming in Python - a different paradigm that complements OOP. You will learn: first-class and higher-order functions, closures and the LEGB scope rule, map, filter, and itertools, functools.partial and functools.reduce, generator functions and expressions, lazy evaluation, and how functional and OOP styles combine in production Python code.

Understanding both paradigms and knowing when to use each is what distinguishes an engineer from a programmer.