Observer Example: An Order Placed Event

Placing an order emails the customer, updates stock and records a sale, and the order code knows none of them.

An order being placed is a natural Observer, and this example maps the template onto it. The order service is the subject. Three observers subscribe to it: one sends a confirmation email, one reduces the stock, and one keeps a running total of takings. The service passes each of them the order, the push form, so none needs to ask for it.

The template's participants in the order example
Template	Order example
Subject	`OrderService`, with `subscribe` and `place`
Observer	`OrderObserver`, a `Protocol` with one method, `order_placed(order)`
ConcreteObserver	`ConfirmationEmail`, `Stock` and `SalesLedger`
Client	The setup at the bottom, which subscribes the three and places two orders

from dataclasses import dataclass
from typing import Protocol


@dataclass(frozen=True)
class Order:
    number: int
    email: str
    items: dict[str, int]
    total_pence: int


class OrderObserver(Protocol):
    def order_placed(self, order: Order) -> None: ...


class OrderService:
    def __init__(self) -> None:
        self._observers: list[OrderObserver] = []

    def subscribe(self, observer: OrderObserver) -> None:
        self._observers.append(observer)

    def place(self, order: Order) -> None:
        print(f"order {order.number} placed")
        for observer in self._observers:
            observer.order_placed(order)


class ConfirmationEmail:
    def order_placed(self, order: Order) -> None:
        print(f"  email: confirmation sent to {order.email}")


class Stock:
    def __init__(self, levels: dict[str, int]) -> None:
        self.levels = levels

    def order_placed(self, order: Order) -> None:
        for sku, quantity in order.items.items():
            self.levels[sku] -= quantity
        print(f"  stock: {self.levels}")


class SalesLedger:
    def __init__(self) -> None:
        self.takings_pence = 0

    def order_placed(self, order: Order) -> None:
        self.takings_pence += order.total_pence
        print(f"  ledger: takings now {self.takings_pence} pence")


orders = OrderService()
orders.subscribe(ConfirmationEmail())
orders.subscribe(Stock({"mug": 10, "tea": 25}))
orders.subscribe(SalesLedger())
orders.place(Order(1001, "[email protected]", {"mug": 2}, 1600))
orders.place(Order(1002, "[email protected]", {"mug": 1, "tea": 3}, 2050))

OutputRan with python:3.12-slim (Python 3.12.13)

order 1001 placed
  email: confirmation sent to [email protected]
  stock: {'mug': 8, 'tea': 25}
  ledger: takings now 1600 pence
order 1002 placed
  email: confirmation sent to [email protected]
  stock: {'mug': 7, 'tea': 22}
  ledger: takings now 3650 pence

Two orders, and every subscriber reacts to each.

After the first order the stock holds 8 mugs and the takings are 1600 pence; after the second, 7 mugs, 22 boxes of tea and 3650 pence. OrderService imports none of the three classes, and its place method would not change if a loyalty-points observer joined them.

The observers are typed with a Protocol, an interface that a class satisfies by having the right method, without inheriting from it. Each observer keeps its own state, the stock levels and the takings, which is why they are classes here rather than functions. In a larger system the same shape goes by the name of an event: order_placed is the event, and the observers are its handlers.

← → move through sections and cards