Python Design Patterns — Singleton, Factory & Observer Tutorial

The Story That Explains Design Patterns

A design pattern is not a library, a class, or code you can copy. It is a template — a way of arranging objects and responsibilities that solves a recurring problem. The pattern's real value is a shared vocabulary between developers and a proven structure that avoids common mistakes.

The Three Families of Patterns

Every classical design pattern falls into one of three categories, based on what it helps you organise: object creation, object composition, or object communication.

In this tutorial we cover three of the most common Python patterns — one from each family (with a heavy focus on Creational and Behavioural since those matter most in Python): Singleton (creational), Factory (creational), and Observer (behavioral).

The Singleton Pattern — One Instance to Rule Them All

A Singleton restricts a class so it can be instantiated only once. All subsequent constructor calls return the existing instance. Used when a resource is naturally global: application configuration, a connection pool, a logger, a cache, an event bus.

Four Ways to Implement Singleton in Python

Python is unusually flexible here. The classical Java-style implementation is one of at least four idiomatic approaches. Each has trade-offs.

Way 1 — Module-Level Singleton (Pythonic Default)

# file: config.py — this file IS the singleton
import os

class _Config:
    def __init__(self):
        self.env      = os.getenv("APP_ENV", "dev")
        self.db_url   = os.getenv("DB_URL", "sqlite:///./app.db")
        self.debug    = self.env == "dev"

# A single instance, created at import time
config = _Config()

# ── Anywhere else in the app ──────────────────────────
from config import config
print(config.db_url)   # same object every time, everywhere

Way 2 — Override __new__

class Logger:
    _instance = None

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

    def __init__(self, name="app"):
        # Careful — __init__ runs on EVERY call, not just the first!
        if not hasattr(self, "_ready"):
            self.name  = name
            self.lines = []
            self._ready = True

    def log(self, msg): self.lines.append(msg)

a = Logger("first")
b = Logger("second")
print(a is b)      # True
print(a.name)      # "first" — __init__ was guarded

Way 3 — Decorator-Based Singleton

from functools import wraps

def singleton(cls):
    instances = {}

    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class Cache:
    def __init__(self):
        self.store = {}

    def set(self, k, v): self.store[k] = v
    def get(self, k):    return self.store.get(k)

c1 = Cache(); c1.set("user:1", "Alice")
c2 = Cache()
print(c2.get("user:1"))   # "Alice" — same store
print(c1 is c2)             # True

Way 4 — Metaclass-Based Singleton

class SingletonMeta(type):
    _instances = {}

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

class DBPool(metaclass=SingletonMeta):
    def __init__(self, size=10):
        self.size        = size
        self.connections = [f"conn-{i}" for i in range(size)]

p1 = DBPool(size=20)
p2 = DBPool(size=99)  # args ignored — instance already exists
print(p1 is p2, p2.size)   # True 20

Singleton — A Thread-Safe Real-World Example

In production, singletons often initialize expensive resources — a database pool, a redis client, a machine-learning model. If two threads both call the constructor at the same time during startup, a naive singleton creates two instances. The fix is a simple lock.

import threading
from typing import ClassVar

class DatabasePool:
    _instance: ClassVar["DatabasePool | None"] = None
    _lock:     ClassVar[threading.Lock]              = threading.Lock()

    def __new__(cls, *args, **kwargs):
        # Double-checked locking — fast path avoids lock after init
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, dsn="postgres://localhost/app", size=10):
        if getattr(self, "_ready", False):
            return
        self.dsn         = dsn
        self.size        = size
        self.connections = [f"conn-{i} @ {dsn}" for i in range(size)]
        self._ready     = True

    def acquire(self):  return self.connections.pop()
    def release(self, c): self.connections.append(c)


# ── Usage from multiple threads ────────────────────
def worker(i):
    pool = DatabasePool()
    conn = pool.acquire()
    print(f"[T{i}] uses {conn}")
    pool.release(conn)

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads: t.start()
for t in threads: t.join()

A Single Shared Database Connection (SQLite)

The pool example above shares many connections. The most classic Singleton use-case is actually simpler — one process, one persistent database connection, reused by every module. This avoids the cost of opening a new TCP handshake or SQLite file handle for every query, and guarantees that transactions and settings (like SQLite's PRAGMA) apply globally.

import sqlite3
import threading
from typing import ClassVar

class Database:
    """Singleton wrapper around a single sqlite3.Connection."""

    _instance:   ClassVar["Database | None"] = None
    _lock:       ClassVar[threading.Lock]         = threading.Lock()

    def __new__(cls, db_path: str = "app.db"):
        # Double-checked locking — fast path avoids the lock after init
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, db_path: str = "app.db"):
        # Guard: __init__ runs on every call, but we only want it once
        if getattr(self, "_ready", False):
            return
        self.db_path     = db_path
        self._connection = sqlite3.connect(
            db_path,
            check_same_thread=False,     # share across threads
            isolation_level=None,        # autocommit; use BEGIN manually
        )
        self._connection.row_factory = sqlite3.Row
        self._connection.execute("PRAGMA journal_mode=WAL")
        self._connection.execute("PRAGMA foreign_keys=ON")
        self._ready = True
        print(f"[Database] new connection opened → {db_path}")

    @property
    def connection(self) -> sqlite3.Connection:
        return self._connection

    def execute(self, sql: str, params: tuple = ()):
        return self._connection.execute(sql, params)

    def query(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
        return self.execute(sql, params).fetchall()

    def close(self):
        self._connection.close()
        Database._instance = None          # allow re-init in tests


# ── Module A: creates the schema ──────────────────
db = Database("shop.db")
db.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id       INTEGER PRIMARY KEY,
        email    TEXT UNIQUE NOT NULL,
        credits  INTEGER DEFAULT 0
    )
""")

# ── Module B (somewhere else): inserts users ──────
db_again = Database("IGNORED.db")     # path arg is ignored — same instance
db_again.execute("INSERT OR IGNORE INTO users(email, credits) VALUES (?, ?)",
                 ("alice@example.com", 100))
db_again.execute("INSERT OR IGNORE INTO users(email, credits) VALUES (?, ?)",
                 ("bob@example.com",   50))

# ── Module C (yet another place): reads users ─────
db_third = Database()
print(f"Same instance? {db is db_again is db_third}")

for row in db_third.query("SELECT email, credits FROM users ORDER BY id"):
    print(dict(row))

The Factory Pattern — Objects on Demand

There are three related variants under the "Factory" umbrella, from simplest to most powerful:

Simple Factory — Payment Gateway Example

Imagine an e-commerce app that supports Stripe, PayPal, and Razorpay. Each has its own SDK, but from the checkout code's point of view, "process a payment" is a single operation. A factory hides the differences.

from abc import ABC, abstractmethod

# ── Common interface ──────────────────────────────
class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, amount: float, currency: str) -> dict: ...

# ── Concrete implementations ──────────────────────
class StripeGateway(PaymentGateway):
    def charge(self, amount, currency):
        return {"gateway": "stripe",
                "charge_id": f"ch_{amount:.0f}",
                "status":    "succeeded"}

class PayPalGateway(PaymentGateway):
    def charge(self, amount, currency):
        return {"gateway": "paypal",
                "payment_id": f"PAY-{amount:.0f}",
                "state":      "approved"}

class RazorpayGateway(PaymentGateway):
    def charge(self, amount, currency):
        return {"gateway": "razorpay",
                "order_id":  f"order_{amount:.0f}",
                "captured":  True}

# ── The factory ───────────────────────────────────
class PaymentFactory:
    _registry = {
        "stripe":   StripeGateway,
        "paypal":   PayPalGateway,
        "razorpay": RazorpayGateway,
    }

    @classmethod
    def create(cls, name: str) -> PaymentGateway:
        name = name.lower().strip()
        if name not in cls._registry:
            raise ValueError(f"Unknown gateway: {name}")
        return cls._registry[name]()

    @classmethod
    def register(cls, name: str, gateway_cls: type[PaymentGateway]):
        # Extensibility: plug in new gateways at runtime
        cls._registry[name.lower()] = gateway_cls


# ── Client code — no if/elif chains ───────────────
for provider in ["stripe", "paypal", "razorpay"]:
    gw     = PaymentFactory.create(provider)
    result = gw.charge(1499.00, "INR")
    print(result)

Factory Method — When Creators Come in Families

The Factory Method variant is used when the creator class hierarchy matches the product class hierarchy. Each creator subclass produces its own product.

from abc import ABC, abstractmethod

# ── Product hierarchy ─────────────────────────────
class Document(ABC):
    @abstractmethod
    def render(self) -> str: ...

class PDFDocument(Document):
    def render(self): return "%PDF-1.7 ... [binary]"

class HTMLDocument(Document):
    def render(self): return "<html>...</html>"

class MarkdownDocument(Document):
    def render(self): return "# Title\n\nParagraph"

# ── Creator hierarchy — one factory_method per subclass ─
class Report(ABC):
    @abstractmethod
    def factory_method(self) -> Document: ...

    def generate(self) -> str:
        doc = self.factory_method()
        return doc.render()

class MonthlyPDFReport(Report):
    def factory_method(self): return PDFDocument()

class MonthlyHTMLReport(Report):
    def factory_method(self): return HTMLDocument()

# ── Client stays generic ─────────────────────────
for report in [MonthlyPDFReport(), MonthlyHTMLReport()]:
    print(report.generate()[:30])

Factory — Anatomy Diagram

The Observer Pattern — Broadcast Without Coupling

Observer is used everywhere: GUI event handlers, spreadsheet cell dependencies, message buses, Redux/Vuex stores, Django signals, JavaScript's DOM events, RxJS streams. Anywhere you see the words "listener", "subscriber", "handler", or "event bus" — it's Observer under the hood.

Observer — Anatomy Diagram

Observer — Practical Example: Stock Price Alerts

from abc import ABC, abstractmethod
from typing import Protocol

# ── Observer interface ────────────────────────────
class Observer(Protocol):
    def update(self, symbol: str, price: float) -> None: ...

# ── Subject ───────────────────────────────────────
class StockTicker:
    def __init__(self, symbol: str):
        self.symbol             = symbol
        self._price            = 0.0
        self._observers: list[Observer] = []

    def attach(self, obs: Observer):
        self._observers.append(obs)

    def detach(self, obs: Observer):
        self._observers.remove(obs)

    def notify(self):
        for obs in self._observers:
            obs.update(self.symbol, self._price)

    @property
    def price(self): return self._price

    @price.setter
    def price(self, value: float):
        self._price = value
        self.notify()           # auto-broadcast on every change

# ── Concrete observers ────────────────────────────
class ConsoleAlert:
    def update(self, symbol, price):
        print(f"[CONSOLE] {symbol} = ${price:.2f}")

class EmailAlert:
    def __init__(self, email, threshold):
        self.email     = email
        self.threshold = threshold

    def update(self, symbol, price):
        if price >= self.threshold:
            print(f"[EMAIL to {self.email}] {symbol} hit ${price:.2f}")

class TradingBot:
    def __init__(self, buy_below):
        self.buy_below = buy_below

    def update(self, symbol, price):
        if price <= self.buy_below:
            print(f"[BOT] BUY 100 x {symbol} @ ${price:.2f}")

# ── Wiring it up ──────────────────────────────────
aapl = StockTicker("AAPL")
aapl.attach(ConsoleAlert())
aapl.attach(EmailAlert("me@example.com", threshold=200))
aapl.attach(TradingBot(buy_below=180))

for new_price in [175.50, 198.20, 203.10]:
    print(f"\n→ Setting price to ${new_price}")
    aapl.price = new_price

Observer — The Pythonic Callback Version

In Python you can skip the classes entirely. Since functions are first-class objects, the simplest observer implementation is a list of callbacks. This is exactly what Django signals, blinker, and pyee do under the hood.

from collections import defaultdict
from typing import Callable, Any

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

    def on(self, event: str):
        # Use as @bus.on("user.created")
        def decorator(fn: Callable):
            self._handlers[event].append(fn)
            return fn
        return decorator

    def emit(self, event: str, **payload: Any):
        for handler in self._handlers[event]:
            handler(**payload)

# ── Global bus (Singleton-ish, module-level) ─────
bus = EventBus()

# ── Register handlers ────────────────────────────
@bus.on("user.created")
def send_welcome_email(user_id, email, **_):
    print(f"→ welcome email to {email} (user {user_id})")

@bus.on("user.created")
def add_to_crm(user_id, email, **_):
    print(f"→ CRM record created for {email}")

@bus.on("user.created")
def log_analytics(user_id, **_):
    print(f"→ analytics: signup event u_{user_id}")

# ── Emit ──────────────────────────────────────────
bus.emit("user.created", user_id=42, email="alice@example.com")

Singleton vs Factory vs Observer — Side by Side

Patterns Combined — A Small Real System

Real code rarely uses one pattern in isolation. Here's a compact example where all three work together — a Singleton event bus, a Factory that creates notification channels, and Observers listening on the bus.

from abc import ABC, abstractmethod
from collections import defaultdict

# ── SINGLETON: the app-wide event bus ─────────────
class EventBusMeta(type):
    _instances = {}
    def __call__(cls, *a, **kw):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*a, **kw)
        return cls._instances[cls]

class EventBus(metaclass=EventBusMeta):
    def __init__(self):
        self._handlers = defaultdict(list)
    def subscribe(self, event, handler):
        self._handlers[event].append(handler)
    def publish(self, event, **data):
        for h in self._handlers[event]:
            h(**data)

# ── FACTORY: notification channel builder ─────────
class Notifier(ABC):
    @abstractmethod
    def send(self, message: str): ...

class EmailNotifier(Notifier):
    def send(self, message): print(f"[EMAIL] {message}")

class SMSNotifier(Notifier):
    def send(self, message): print(f"[SMS]   {message}")

class SlackNotifier(Notifier):
    def send(self, message): print(f"[SLACK] {message}")

class NotifierFactory:
    _map = {"email": EmailNotifier, "sms": SMSNotifier, "slack": SlackNotifier}
    @classmethod
    def make(cls, channel): return cls._map[channel]()

# ── OBSERVER: subscribers on the bus ──────────────
bus = EventBus()

for channel in ("email", "sms", "slack"):
    notifier = NotifierFactory.make(channel)
    bus.subscribe("order.paid",
                  lambda order_id, amount, n=notifier:
                      n.send(f"Order {order_id} paid: ${amount}"))

# ── Somewhere else in the code ───────────────────
EventBus().publish("order.paid", order_id="ORD-001", amount=149.99)

When to Use Each Pattern

Testing Code That Uses These Patterns

# ── Testable observer with a spy ─────────────────
class SpyObserver:
    def __init__(self): self.received = []
    def update(self, symbol, price):
        self.received.append((symbol, price))

def test_ticker_notifies_all_observers():
    ticker = StockTicker("MSFT")
    spy1, spy2 = SpyObserver(), SpyObserver()
    ticker.attach(spy1)
    ticker.attach(spy2)

    ticker.price = 420.50

    assert spy1.received == [("MSFT", 420.50)]
    assert spy2.received == [("MSFT", 420.50)]

def test_factory_creates_correct_gateway():
    assert isinstance(PaymentFactory.create("stripe"), StripeGateway)
    assert isinstance(PaymentFactory.create("paypal"), PayPalGateway)

Golden Rules