Behavioral Design Patterns: Strategy, Observer and Command in Python

The shipping rule for an online shop changed four times in one quarter: flat rate, then weight-based, then free above a threshold, then express for certain cities. Each change meant editing the checkout function and re-testing everything else in it. The rule that keeps changing is the part to pull out, and giving it a name and a shape is what behavioral patterns do: they describe how objects divide responsibility and talk to each other.

Strategy: make the changing rule a parameter

A strategy is an interchangeable algorithm behind a fixed signature. The code that needs the algorithm (here, Checkout) receives it from outside and calls it, without knowing which one it has. In languages without first-class functions that means an interface plus one class per algorithm. In Python a plain function with the agreed signature is the strategy, and anything that needs configuration can be a closure that returns one.

from dataclasses import dataclass


@dataclass
class Parcel:
    weight_kg: float
    distance_km: int


# Each strategy is just a function from Parcel to price. Same signature = interchangeable.
def standard(parcel):
    return 50 + 12 * parcel.weight_kg


def express(parcel):
    return 120 + 20 * parcel.weight_kg + 0.5 * parcel.distance_km


def free_above(threshold, inner):
    """A strategy that carries its own settings and wraps another strategy."""
    def strategy(parcel):
        price = inner(parcel)
        return 0 if price >= threshold else price
    return strategy


class Checkout:
    def __init__(self, shipping):
        self.shipping = shipping               # the strategy is injected, not hard-coded

    def total(self, items_price, parcel):
        return items_price + self.shipping(parcel)


parcel = Parcel(weight_kg=4, distance_km=300)
for name, strategy in [
    ("standard", standard),
    ("express", express),
    ("standard, free from 90", free_above(90, standard)),
    ("express, free from 400", free_above(400, express)),
]:
    print(f"{name:24} -> {Checkout(strategy).total(1500, parcel):.2f}")

checkout = Checkout(standard)
print(f"before: {checkout.total(1500, parcel):.2f}")
checkout.shipping = express                    # swap behaviour at run time
print(f"after:  {checkout.total(1500, parcel):.2f}")

standard                 -> 1598.00
express                  -> 1850.00
standard, free from 90   -> 1500.00
express, free from 400   -> 1850.00
before: 1598.00
after:  1850.00

Four strategies run through the same Checkout, and the last lines swap one in at run time. free_above shows the second advantage over an if chain: strategies compose. It takes another strategy as input and wraps it, so "standard shipping, free from 90" and "express shipping, free from 400" need no new classes. Strategy is the same idea as the injected Renderer and PricingRule objects in the SOLID refactor, and that is not an accident: dependency injection is how strategies usually arrive.

Use a strategy when you have several ways of doing one thing and the choice depends on context or configuration. Do not use it for two branches that will never grow; an if is clearer.

Observer: tell everyone who cares, without knowing who they are

The observer pattern lets one object, the subject, notify a changing list of listeners when something happens, without depending on any of them. A build server finishing a job should update a dashboard and ping chat on failure, but the build code must not import the dashboard or the chat client. Listeners subscribe, the subject publishes, and the subject's code never changes when a new listener arrives.

class BuildEvents:
    """The subject: holds a list of listeners and tells them when something happens."""

    def __init__(self):
        self._listeners = []

    def subscribe(self, listener):
        self._listeners.append(listener)
        return lambda: self._listeners.remove(listener)      # returns an "unsubscribe" handle

    def publish(self, event):
        for listener in list(self._listeners):               # copy: listeners may unsubscribe
            try:
                listener(event)
            except Exception as err:                         # one bad listener must not block the rest
                print(f"  (listener {listener.__name__} failed: {err})")


def dashboard(event):
    print(f"  dashboard: {event['job']} is {event['status']}")


def chat_notifier(event):
    if event["status"] == "failed":
        print(f"  chat: @oncall {event['job']} failed")


def flaky_audit_log(event):
    raise RuntimeError("disk full")


bus = BuildEvents()
unsubscribe_dashboard = bus.subscribe(dashboard)
bus.subscribe(flaky_audit_log)
bus.subscribe(chat_notifier)

print("event 1")
bus.publish({"job": "api-tests", "status": "passed"})
print("event 2")
bus.publish({"job": "web-build", "status": "failed"})

unsubscribe_dashboard()
print("event 3 (dashboard unsubscribed)")
bus.publish({"job": "web-build", "status": "passed"})

event 1
  dashboard: api-tests is passed
  (listener flaky_audit_log failed: disk full)
event 2
  dashboard: web-build is failed
  (listener flaky_audit_log failed: disk full)
  chat: @oncall web-build failed
event 3 (dashboard unsubscribed)
  (listener flaky_audit_log failed: disk full)

Three details in this small implementation are what separate a toy from something safe. publish iterates over a copy of the list, so a listener can unsubscribe itself during notification without skipping the next one. Each call is wrapped so that flaky_audit_log raising an exception is reported but does not stop the chat notifier from running. And subscribe returns an unsubscribe handle, which makes cleaning up the caller's responsibility and is easy to do correctly.

This version is synchronous and in-process: publish returns only after every listener has run, and a slow listener slows the publisher. When listeners live in other services or must not block the publisher, the same idea becomes a message queue or event stream, covered in the Apache Kafka course.

Command: turn an action into an object

A command packages a request, together with everything needed to carry it out, as an object. That sounds like a roundabout way to call a function, until you need something a plain call cannot give you: undo, redo, queueing, scheduling, retrying or an audit log of what was done. Each command knows how to do() itself and how to undo() its own effect.

class AddStock:
    def __init__(self, inventory, sku, qty):
        self.inventory, self.sku, self.qty = inventory, sku, qty

    def do(self):
        self.inventory[self.sku] = self.inventory.get(self.sku, 0) + self.qty

    def undo(self):
        self.inventory[self.sku] -= self.qty

    def __str__(self):
        return f"add {self.qty} x {self.sku}"


class RemoveStock:
    def __init__(self, inventory, sku, qty):
        self.inventory, self.sku, self.qty = inventory, sku, qty

    def do(self):
        if self.inventory.get(self.sku, 0) < self.qty:
            raise ValueError(f"not enough {self.sku}")
        self.inventory[self.sku] -= self.qty

    def undo(self):
        self.inventory[self.sku] += self.qty

    def __str__(self):
        return f"remove {self.qty} x {self.sku}"


class History:
    def __init__(self):
        self._done, self._undone = [], []

    def run(self, command):
        command.do()                       # if do() raises, nothing is recorded
        self._done.append(command)
        self._undone.clear()               # a new action invalidates the redo path

    def undo(self):
        if self._done:
            command = self._done.pop()
            command.undo()
            self._undone.append(command)
            print("undid:", command)

    def redo(self):
        if self._undone:
            command = self._undone.pop()
            command.do()
            self._done.append(command)
            print("redid:", command)

    def log(self):
        return [str(c) for c in self._done]


stock = {"bolt": 100}
history = History()
history.run(AddStock(stock, "bolt", 50))
history.run(RemoveStock(stock, "bolt", 120))
print(stock, history.log())

history.undo()
print(stock)
history.redo()
print(stock)

try:
    history.run(RemoveStock(stock, "bolt", 999))
except ValueError as err:
    print("rejected:", err, "| history unchanged:", history.log())

{'bolt': 30} ['add 50 x bolt', 'remove 120 x bolt']
undid: remove 120 x bolt
{'bolt': 150}
redid: remove 120 x bolt
{'bolt': 30}
rejected: not enough bolt | history unchanged: ['add 50 x bolt', 'remove 120 x bolt']

Follow the output. Two commands run and the stock is 30. undo reverses the removal (150), redo re-applies it (30). The final failing command raises inside run before it is recorded, so the history is unchanged: a rejected action must not appear in the log or it could never be undone correctly. Note also _undone.clear() in run: performing a fresh action invalidates the redo path, which is how every editor behaves.

Common mistakes

• A strategy class with a single method and no state. In Python a function does the job with less code. Reserve classes for strategies that carry real configuration or several related methods.
• Strategies with different signatures. If express needs the destination city and standard does not, the caller starts type-checking again. Pass one parameter object that holds everything any strategy might need.
• Mutating the listener list while notifying. Removing from self._listeners inside the loop raises no error but silently skips the next listener. Iterate over a copy.
• One failing observer breaks the rest. Without the try/except, the exception from flaky_audit_log would propagate out of publish and the chat message would never be sent.
• Forgetting to unsubscribe. Objects stay alive because the subject references them; memory grows and "dead" screens keep reacting to events.
• Assuming listener order. Observers should be independent. If one must run before another, the design is really a pipeline.
• Recording a command before it succeeds. The history then contains actions that never happened, and undo fails or corrupts state.
• Undo that does not restore exactly. undo for a removal that also triggered side effects (an email sent) cannot be a true inverse. Decide what is undoable and say so.

How this shows up in real projects and interviews

Strategy is the quiet workhorse: payment providers, pricing rules, retry policies, sorting comparators (sorted(key=...) takes a strategy). Observer appears as event listeners in UI frameworks, signals in web frameworks, and webhooks at system scale. Command underlies undo stacks, job queues, database migration steps and the "action" objects of state-management libraries.

Interviews often hand you a messy if/elif chain and ask how to refactor it, hoping for strategy, or ask you to design a text editor's undo, hoping for command with two stacks. Say the trade-off aloud: more objects and indirection in exchange for a rule that changes without touching its caller. The final lesson, choosing and avoiding design patterns, covers when not to pay that price.