Chain of Responsibility (handlers pass the request along)

When to use

• You have a request/record that should flow through handlers in order: validate → enrich → mask → route.
• Some handlers may stop early (reject/drop) or modify the request.
• You want to add/remove/reorder steps without changing each handler.

Avoid when you only need a straight list of pure functions (then just loop).

Diagram (text)

Event ─→ Validate ─→ MaskPII ─→ Router ─→ (done)
          │(raise on bad)       │(side effect: writes to sink)

Python example (≤40 lines, type-hinted)

from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol, Optional, Dict, List

Event = Dict[str, object]

class Handler(Protocol):
    def set_next(self, nxt: "Handler") -> "Handler": ...
    def handle(self, e: Event) -> Event: ...

@dataclass
class BaseHandler:
    nxt: Optional[Handler] = None
    def set_next(self, nxt: Handler) -> Handler: self.nxt = nxt; return nxt
    def handle(self, e: Event) -> Event: return self.nxt.handle(e) if self.nxt else e

class Validate(BaseHandler):
    def handle(self, e: Event) -> Event:
        if "user" not in e or "email" not in e: raise ValueError("invalid event")
        return super().handle(e)

class MaskPII(BaseHandler):
    def handle(self, e: Event) -> Event:
        email = str(e["email"]); user, _, dom = email.partition("@")
        e["email"] = "***@" + dom if dom else "***"
        return super().handle(e)

@dataclass
class Router(BaseHandler):
    sink: Dict[str, List[Event]]
    def handle(self, e: Event) -> Event:
        self.sink.setdefault(str(e.get("type","other")), []).append(dict(e))
        return super().handle(e)

# build the chain
def build_chain(sink: Dict[str, List[Event]]) -> Handler:
    head = Validate(); head.set_next(MaskPII()).set_next(Router(sink))
    return head

Tiny pytest (cements it)

def test_chain_runs_and_masks():
    sink = {}
    chain = build_chain(sink)
    evt = {"type":"signup","user":"u1","email":"a@b.com"}
    out = chain.handle(evt)
    assert sink["signup"][0]["email"].startswith("***@")
    assert out is evt  # same object flowing through

def test_chain_stops_on_bad_event():
    sink = {}; chain = build_chain(sink)
    import pytest
    with pytest.raises(ValueError): chain.handle({"type":"signup","email":"x@y"})
    assert sink == {}  # router not reached

Trade-offs & pitfalls

• Pros: Pluggable steps; simple add/remove/reorder; each handler focused; easy to test in isolation.
• Cons: Control flow is implicit; debugging order can be tricky.
• Pitfalls:
  • Forgetting to call super().handle(e) → chain breaks silently.
  • Hidden side effects—document what each handler reads/writes.
  • Overusing exceptions for control flow—prefer clear returns when not an error.

Pythonic alternatives

• Plain function pipeline: for fn in steps: e = fn(e) (best when no early-stop/branching).
• Middleware style (ASGI/WSGI): handlers accept (event, next) and decide whether to call next().
• Composite if you just need grouping, not early-stop.
• Decorator for cross-cutting concerns around a single handler.

Mini exercise

Add a DropPII handler that removes the email field entirely when e.get("type") == "internal", then continues. Insert it between Validate and MaskPII. Write a test proving internal events reach Router without any email key.

Checks (quick checklist)

• Each handler does one job and then super().handle(e).
• Order is explicit and tested.
• Early-stop behavior is clear (raise or return).
• Handlers don’t depend on each other’s internals—only on the event.
• Easy to add/remove/replace a handler without touching others.

Say “next” and we’ll do #20 Mediator (a coordinator that reduces many-to-many chatter between components, e.g., an orchestrator and multiple workers).