If you’ve ever written code that constantly checks “has anything changed yet?” on a loop, or copy-pasted the same notification logic into ten different places every time something happens in your app, this post is for you.
The Observer pattern flips the script: instead of callers asking “did anything change?”, the thing that changed tells everyone who cares. Subscribers register their interest upfront. When the event fires, they all get notified automatically. No polling. No tight coupling. No missed updates.
The Problem: Everyone Wants to Know When the Price Moves
Say you’re building a stock alert system. When a stock price changes, several things need to happen, for instance we need to send an email alert, log the change to an audit trail, trigger a portfolio rebalance check. The naive approach wires all of that directly into the price update logic:
# BEFORE: Everything hardcoded into the price update function
import smtplib
import logging
logger = logging.getLogger(__name__)
def update_stock_price(ticker: str, new_price: float, subscribers: list):
print(f"{ticker} price updated to ${new_price:.2f}")
# Send email alerts to all subscribers
for subscriber in subscribers:
with smtplib.SMTP("smtp.gmail.com", 587) as smtp:
smtp.sendmail(
"alerts@myapp.com",
subscriber.email,
f"Subject: {ticker} Alert\n\n{ticker} is now ${new_price:.2f}"
)
# Log to audit trail
logger.info(f"PRICE_CHANGE | {ticker} | ${new_price:.2f}")
# Check portfolio rebalance
for subscriber in subscribers:
if subscriber.portfolio.needs_rebalance(ticker, new_price):
subscriber.portfolio.rebalance(ticker, new_price)
This is already messy, and the system isn’t even finished yet. The problems:
- Price update logic is doing too many jobs: it shouldn’t know anything about SMTP servers, logging formats, or portfolio rebalancing. Its job is to update a price.
- Adding new reactions means editing this function: want to add a push notification or a Slack alert? You’re cracking open
update_stock_priceevery time, risking breaking what already works. - No way to subscribe or unsubscribe dynamically: a user who turns off email alerts still gets them, because the subscriber list is baked in.
- Impossible to test in isolation: unit testing
update_stock_pricemeans dealing with live SMTP connections, real loggers, and portfolio objects all at once.
What you want is a system where update_stock_price does exactly one thing — update the price — and anything that cares about that event registers itself and handles its own reaction.
The Fix: A Stock Ticker That Notifies Its Observers
# AFTER: Observer pattern for stock price alerts
from abc import ABC, abstractmethod
from typing import List
# --- Step 1: The Observer interface ---
class StockObserver(ABC):
@abstractmethod
def update(self, ticker: str, new_price: float) -> None:
pass
# --- Step 2: Concrete observers, each handling their own concern ---
class EmailAlertObserver(StockObserver):
def __init__(self, email: str, threshold: float):
self.email = email
self.threshold = threshold
def update(self, ticker: str, new_price: float) -> None:
if new_price <= self.threshold:
print(f"[EMAIL] Sending alert to {self.email}: "
f"{ticker} dropped to ${new_price:.2f} (threshold: ${self.threshold:.2f})")
class AuditLogObserver(StockObserver):
def update(self, ticker: str, new_price: float) -> None:
print(f"[AUDIT LOG] {ticker} | new price: ${new_price:.2f}")
class PortfolioRebalanceObserver(StockObserver):
def __init__(self, owner: str):
self.owner = owner
def update(self, ticker: str, new_price: float) -> None:
print(f"[PORTFOLIO] Checking rebalance for {self.owner} "
f"after {ticker} moved to ${new_price:.2f}")
# --- Step 3: The subject — the stock ticker itself ---
class StockTicker:
def __init__(self, ticker: str):
self.ticker = ticker
self._price: float = 0.0
self._observers: List[StockObserver] = []
def subscribe(self, observer: StockObserver) -> None:
self._observers.append(observer)
def unsubscribe(self, observer: StockObserver) -> None:
self._observers.remove(observer)
def set_price(self, new_price: float) -> None:
self._price = new_price
print(f"\n{self.ticker} updated to ${new_price:.2f}")
self._notify_observers(new_price)
def _notify_observers(self, new_price: float) -> None:
for observer in self._observers:
observer.update(self.ticker, new_price)
# --- Usage ---
aapl = StockTicker("AAPL")
aapl.subscribe(EmailAlertObserver(email="alice@example.com", threshold=170.00))
aapl.subscribe(EmailAlertObserver(email="bob@example.com", threshold=165.00))
aapl.subscribe(AuditLogObserver())
aapl.subscribe(PortfolioRebalanceObserver(owner="Alice"))
aapl.set_price(168.50)
# → [EMAIL] Sending alert to alice@example.com: AAPL dropped to $168.50
# → [AUDIT LOG] AAPL | new price: $168.50
# → [PORTFOLIO] Checking rebalance for Alice after AAPL moved to $168.50
# Bob's threshold not hit, so no email for him — each observer decides for itself
# Alice turns off alerts
aapl.unsubscribe(aapl._observers[0])
aapl.set_price(163.00)
# → [AUDIT LOG] AAPL | new price: $163.00 (no email to Alice anymore)
StockTicker now does exactly one thing — track a price and tell its observers when it changes. It has no idea what those observers do. EmailAlertObserver could be replaced, removed, or multiplied without touching a single line of StockTicker.
Adding a Slack notification? Write a SlackAlertObserver, call aapl.subscribe(), done. No existing code changes.
Observer vs. Factory: What’s the Difference?
After reading the Factory post, you might wonder where these two patterns overlap — both involve a list of implementations behind a shared interface. They’re actually solving completely different problems.
Factory is about creation. You have multiple types of things and need to decide which one to build at a given moment. The factory makes that decision and hands you one object. It’s a one-time construction choice.
Observer is about communication. You have one thing that changes, and multiple other things that need to react to that change. The subject doesn’t choose who gets notified — everyone subscribed gets the event. It’s an ongoing broadcast relationship.
A concrete way to see the difference: in the Factory post, NotificationFactory.get_sender("email") gives you one sender for one task. In this post, aapl.set_price(168.50) notifies all observers simultaneously — email, audit log, portfolio check all fire at once.
They can also work together naturally. You could use a Factory to create the right observer type and then subscribe it:
# Factory creates the observer, Observer pattern handles the event
observer = AlertObserverFactory.get_observer(user.alert_preference)
aapl.subscribe(observer)
Quick rule of thumb: Factory answers “what should I build?” Observer answers “who needs to know about this?”
What About Thread Safety?
In a live trading system, prices update fast and potentially from multiple threads. The _observers list in StockTicker is a shared mutable object — if one thread is iterating over it while another calls subscribe() or unsubscribe(), you can get a RuntimeError: list changed size during iteration.
A simple fix:
import threading
from typing import List
class StockTicker:
def __init__(self, ticker: str):
self.ticker = ticker
self._price: float = 0.0
self._observers: List[StockObserver] = []
self._lock = threading.Lock()
def subscribe(self, observer: StockObserver) -> None:
with self._lock:
self._observers.append(observer)
def unsubscribe(self, observer: StockObserver) -> None:
with self._lock:
self._observers.remove(observer)
def _notify_observers(self, new_price: float) -> None:
with self._lock:
observers_snapshot = list(self._observers) # copy to iterate safely
for observer in observers_snapshot:
observer.update(self.ticker, new_price)
The snapshot trick is important — take a copy of the list inside the lock, then iterate the copy outside it. This prevents blocking the entire notification loop while observers do their (potentially slow) work, like sending an email or writing to a database.
Python’s GIL gives you some protection, but for anything involving I/O or thread pool executors, the explicit lock is the right call.
Other Places Observers Earn Their Keep
Once you see the pattern, it’s everywhere:
Django signals Django’s post_save, pre_delete, and custom signals are the Observer pattern baked into the framework. A model saves, and every connected signal handler fires automatically — send a welcome email, invalidate a cache, write to an analytics table — all without the model knowing any of those things exist.
Event-driven UIs Every button.addEventListener("click", handler) in JavaScript is the Observer pattern. Python GUI frameworks like tkinter and PyQt use the same idea — widgets emit events, handlers subscribe to them.
Webhooks When a payment processor, GitHub, or Stripe sends a POST request to your endpoint after something happens, that’s an Observer pattern over HTTP. Your app subscribed to an event; the external service is the subject notifying you.
The Takeaway
The Observer pattern is the right tool when:
- One object changes state and an unknown or variable number of other objects need to react
- You want those reactions to be decoupled — the subject shouldn’t know what its observers do
- You need to add, remove, or swap reactions at runtime without changing the thing being observed
Don’t reach for it when you have a simple one-to-one relationship with no need for dynamic subscription — that’s just a function call. But the moment you find yourself hardcoding multiple reactions into a single event, or polling for changes on a timer, the Observer pattern is almost certainly what you need.
One subject. Many listeners. Zero tight coupling. That’s the whole idea.