Intermediate Examples

"""Example 28: An Abstract Factory for UI Widget Families."""  # => module docstring
 
import abc  # => imports the abc module
 
 
class Button(abc.ABC):  # => the ABSTRACT product: every theme has SOME Button
    @abc.abstractmethod  # => marks the next method as required for every subclass
    def paint(self) -> str:  # => no body -- a required contract for subclasses
        ...  # => the ellipsis stub -- concrete Buttons below fill this in
 
 
class Checkbox(abc.ABC):  # => the ABSTRACT product: every theme has SOME Checkbox
    @abc.abstractmethod  # => marks the next method as required for every subclass
    def paint(self) -> str:  # => no body -- a required contract for subclasses
        ...  # => the ellipsis stub -- concrete Checkboxes below fill this in
 
 
class DarkButton(Button):  # => a CONCRETE product belonging to the dark family
    def paint(self) -> str:  # => defines the paint() method
        return "dark-button"  # => tags this product with its family name
 
 
class DarkCheckbox(Checkbox):  # => a CONCRETE product belonging to the dark family
    def paint(self) -> str:  # => defines the paint() method
        return "dark-checkbox"  # => tags this product with its family name
 
 
class LightButton(Button):  # => a CONCRETE product belonging to the light family
    def paint(self) -> str:  # => defines the paint() method
        return "light-button"  # => tags this product with its family name
 
 
class LightCheckbox(Checkbox):  # => a CONCRETE product belonging to the light family
    def paint(self) -> str:  # => defines the paint() method
        return "light-checkbox"  # => tags this product with its family name
 
 
class WidgetFactory(abc.ABC):  # => the ABSTRACT FACTORY -- produces a MATCHED family
    @abc.abstractmethod  # => marks the next method as required for every WidgetFactory subclass
    def create_button(self) -> Button:  # => no body -- required for every factory
        ...  # => the ellipsis stub -- concrete factories below fill this in
 
    @abc.abstractmethod  # => marks the next method as required for every WidgetFactory subclass
    def create_checkbox(self) -> Checkbox:  # => no body -- required for every factory
        ...  # => the ellipsis stub -- concrete factories below fill this in
 
 
class DarkThemeFactory(WidgetFactory):  # => produces ONLY dark-family products
    def create_button(self) -> Button:  # => defines the create_button() method
        return DarkButton()  # => always matches the dark family, never mixed
 
    def create_checkbox(self) -> Checkbox:  # => defines the create_checkbox() method
        return DarkCheckbox()  # => always matches the dark family, never mixed
 
 
class LightThemeFactory(WidgetFactory):  # => produces ONLY light-family products
    def create_button(self) -> Button:  # => defines the create_button() method
        return LightButton()  # => always matches the light family, never mixed
 
    def create_checkbox(self) -> Checkbox:  # => defines the create_checkbox() method
        return LightCheckbox()  # => always matches the light family, never mixed
 
 
def build_toolbar(  # => signature spans multiple lines: parameter below, return type on the closing line
    factory: WidgetFactory,  # => the single parameter -- typed as the ABSTRACT WidgetFactory, never a concrete subclass
) -> tuple[str, str]:  # => the ONE function every caller uses, regardless of theme
    # => the client depends ONLY on WidgetFactory -- it never names a concrete class
    button: Button = factory.create_button()  # => whichever family the factory picks
    checkbox: Checkbox = factory.create_checkbox()  # => the SAME family, guaranteed
    return button.paint(), checkbox.paint()  # => returns this value to the caller
 
 
dark_ui: tuple[str, str] = build_toolbar(DarkThemeFactory())  # => swap the ONE argument
light_ui: tuple[str, str] = build_toolbar(LightThemeFactory())  # => whole family swaps
print(dark_ui)  # => both products come from the dark family together
# => Output: ('dark-button', 'dark-checkbox')
print(light_ui)  # => both products come from the light family together
# => Output: ('light-button', 'light-checkbox')
# => Swapping ONE factory argument swaps the ENTIRE matched product family at once

('dark-button', 'dark-checkbox')
('light-button', 'light-checkbox')

"""Example 28: pytest verification for An Abstract Factory for UI Widget Families."""
 
from example import DarkThemeFactory, LightThemeFactory, build_toolbar
 
 
def test_dark_factory_produces_only_dark_family_products() -> None:
    button, checkbox = build_toolbar(DarkThemeFactory())
    assert button == "dark-button"  # => both products share the dark family
    assert checkbox == "dark-checkbox"  # => never mixed with the light family
 
 
def test_swapping_the_factory_swaps_the_whole_family() -> None:
    button, checkbox = build_toolbar(LightThemeFactory())
    assert (button, checkbox) == (
        "light-button",
        "light-checkbox",
    )  # => one swapped argument, whole family changed
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 29: A Fluent Builder for HTTP Requests."""
 
from dataclasses import dataclass  # => imports dataclass from dataclasses
 
 
@dataclass(frozen=True)  # => the finished product -- immutable once built() returns it
class Request:  # => begins the Request class body
    method: str  # => a required field, part of the generated __init__
    url: str  # => a required field, part of the generated __init__
    headers: dict[str, str]  # => a required field, part of the generated __init__
    body: str | None  # => a required field, part of the generated __init__
 
 
class RequestBuilder:  # => begins the RequestBuilder class body
    def __init__(self, url: str) -> None:  # => only the ONE truly required piece
        self._url: str = url  # => stores the required url on this instance
        self._method: str = "GET"  # => a sensible default -- no need to always name it
        self._headers: dict[str, str] = {}  # => starts empty, grows via with_header
        self._body: str | None = None  # => optional -- most requests have no body
 
    def with_method(self, method: str) -> "RequestBuilder":  # => fluent step
        self._method = method  # => mutates this builder's own state
        return self  # => returning self is what enables the NEXT chained call
 
    def with_header(self, key: str, value: str) -> "RequestBuilder":  # => fluent step
        self._headers[key] = value  # => accumulates one header per call
        return self  # => returning self is what enables the NEXT chained call
 
    def with_body(self, body: str) -> "RequestBuilder":  # => fluent step
        self._body = body  # => mutates this builder's own state
        return self  # => returning self is what enables the NEXT chained call
 
    def build(self) -> Request:  # => the terminal step -- assembles the final object
        return Request(self._method, self._url, dict(self._headers), self._body)  # => a fresh, immutable snapshot -- not a live view of the builder
 
 
request: Request = (
    RequestBuilder("https://api.example.com/orders")  # => only the required piece
    .with_method("POST")  # => optional pieces chained in ANY order needed
    .with_header("Content-Type", "application/json")  # => optional pieces
    .with_body('{"item": "widget"}')  # => optional pieces
    .build()  # => terminal call -- returns the finished, immutable Request
)
print(request.method, request.url)  # => confirms the chained values landed correctly
# => Output: POST https://api.example.com/orders
print(request.headers, request.body)  # => confirms headers and body landed correctly
# => Output: {'Content-Type': 'application/json'} {"item": "widget"}
# => A fluent builder assembles optional parts step by step, with no telescoping constructor

POST https://api.example.com/orders
{'Content-Type': 'application/json'} {"item": "widget"}

"""Example 29: pytest verification for A Fluent Builder for HTTP Requests."""
 
from example import Request, RequestBuilder
 
 
def test_chained_builder_assembles_a_complete_request() -> None:
    request: Request = RequestBuilder("https://api.example.com/orders").with_method("POST").with_header("Content-Type", "application/json").with_body('{"item": "widget"}').build()
    assert request.method == "POST"
    assert request.headers == {"Content-Type": "application/json"}
    assert request.body == '{"item": "widget"}'
 
 
def test_omitted_optional_parts_use_sensible_defaults() -> None:
    request: Request = RequestBuilder("https://api.example.com/health").build()
    assert request.method == "GET"  # => no with_method() call, defaults apply
    assert request.headers == {}  # => no with_header() calls at all
    assert request.body is None  # => no with_body() call at all
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 30: A Config Singleton, and the Global-State Seam It Creates."""
 
 
class Config:  # => begins the Config class body
    _instance: "Config | None" = None  # => a CLASS attribute -- shared by every caller
 
    def __init__(self) -> None:  # => never call this directly -- use instance() instead
        self.debug: bool = False  # => the mutable state every caller ends up sharing
 
    @classmethod
    def instance(cls) -> "Config":  # => cls is Config itself, the shared holder
        if cls._instance is None:  # => only the FIRST call actually constructs one
            cls._instance = cls()  # => every later call reuses this same object
        return cls._instance  # => always the SAME object, never a fresh one
 
    @classmethod
    def reset(cls) -> None:  # => the escape hatch this pattern FORCES callers to add
        cls._instance = None  # => without this, no caller can ever start clean again
 
 
first: Config = Config.instance()  # => the first call constructs the shared instance
second: Config = Config.instance()  # => the second call reuses the SAME instance
print(first is second)  # => proves there is only ever one Config in the whole process
# => Output: True
 
first.debug = True  # => mutating through ONE reference...
print(second.debug)  # => ...is visible through EVERY OTHER reference -- global state
# => Output: True
# => `Config.instance()` always returns the SAME object, so a mutation anywhere leaks everywhere
 
Config.reset()  # => the ONLY way to get a clean Config again -- an explicit reset call
fresh: Config = Config.instance()  # => a genuinely NEW instance, post-reset
print(fresh.debug)  # => the leaked True from `first.debug = True` above is gone
# => Output: False
# => Every consumer of a singleton is implicitly coupled to every OTHER consumer through shared state

True
True
False

"""Example 30: pytest verification for A Config Singleton, and Its Cost."""
 
from example import Config
 
 
def test_instance_always_returns_the_same_object() -> None:
    Config.reset()  # => start from a known-clean state -- the cost this test proves
    assert Config.instance() is Config.instance()  # => two calls, one shared object
 
 
def test_mutation_leaks_across_every_reference_without_reset() -> None:
    Config.reset()  # => without this line, a PRIOR test's leftover state leaks in here
    a: Config = Config.instance()
    a.debug = True  # => mutate through one reference
    b: Config = Config.instance()
    assert b.debug is True  # => visible through a totally different reference -- the pain
    Config.reset()  # => clean up so later tests are not poisoned by this test's mutation
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 31: Replacing a Singleton with an Injected Dependency."""
 
from dataclasses import dataclass  # => imports dataclass from dataclasses
 
 
@dataclass  # => a PLAIN, ordinary object now -- no shared instance() anywhere
class Config:  # => begins the Config class body
    debug: bool = False  # => a normal field with a normal default
 
 
class ReportGenerator:  # => begins the ReportGenerator class body
    def __init__(self, config: Config) -> None:  # => the config is INJECTED, never looked up globally
        self.config = config  # => held as an ordinary instance field
 
    def summary(self) -> str:  # => defines the summary() method
        mode: str = "debug" if self.config.debug else "release"  # => reads ONLY its own injected config
        return f"report ({mode})"  # => returns this value to the caller
 
 
prod_config: Config = Config(debug=False)  # => one INDEPENDENT Config instance
debug_config: Config = Config(debug=True)  # => a SECOND, unrelated Config instance
 
prod_report: ReportGenerator = ReportGenerator(prod_config)  # => wired to prod_config
debug_report: ReportGenerator = ReportGenerator(debug_config)  # => wired to debug_config
print(prod_report.summary())  # => reads ONLY prod_config -- unaffected by debug_config
# => Output: report (release)
print(debug_report.summary())  # => reads ONLY debug_config -- unaffected by prod_config
# => Output: report (debug)
# => Two ReportGenerators, each holding its OWN Config, never share state -- no reset() needed anywhere

report (release)
report (debug)

"""Example 31: pytest verification for Replacing a Singleton with an Injected Dependency."""
 
from example import Config, ReportGenerator
 
 
def test_debug_mode_report_is_isolated_from_prod_mode_report() -> None:
    # => no Config.reset() call anywhere -- there is no global state left to reset
    prod: ReportGenerator = ReportGenerator(Config(debug=False))
    debug: ReportGenerator = ReportGenerator(Config(debug=True))
    assert prod.summary() == "report (release)"  # => unaffected by the debug instance
    assert debug.summary() == "report (debug)"  # => unaffected by the prod instance
 
 
def test_two_instances_never_share_underlying_state() -> None:
    a: Config = Config(debug=False)
    b: Config = Config(debug=False)
    a.debug = True  # => mutate ONLY a's config
    assert b.debug is False  # => b is a genuinely separate object -- no leak at all
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 32: A Virtual Proxy Defers an Expensive Load."""
 
 
class RealImage:  # => the EXPENSIVE real subject -- loading it costs real work
    load_count: int = 0  # => a class-level counter so the example can OBSERVE loading
 
    def __init__(self, filename: str) -> None:  # => construction itself IS the expensive step
        self.filename = filename  # => stores filename on this instance
        RealImage.load_count += 1  # => every construction increments the shared counter
        print(f"loading {filename} from disk")  # => simulates the expensive I/O happening
 
    def render(self) -> str:  # => defines the render() method
        return f"rendering {self.filename}"  # => returns this value to the caller
 
 
class ImageProxy:  # => the VIRTUAL PROXY -- stands in for a RealImage not yet loaded
    def __init__(self, filename: str) -> None:  # => cheap -- no loading happens here
        self._filename: str = filename  # => stores only the filename, nothing expensive
        self._real: RealImage | None = None  # => the real subject, created LAZILY
 
    def render(self) -> str:  # => same interface as RealImage.render() -- callers can't tell
        if self._real is None:  # => only construct the real subject on FIRST access
            self._real = RealImage(self._filename)  # => the expensive load happens HERE
        return self._real.render()  # => delegates to the now-loaded real subject
 
 
proxy: ImageProxy = ImageProxy("photo.png")  # => cheap -- RealImage.load_count is still 0
print(RealImage.load_count)  # => confirms NOTHING loaded yet, just constructing the proxy
# => Output: 0
proxy.render()  # => first access -- triggers the actual, expensive load
proxy.render()  # => second access -- reuses the ALREADY-loaded RealImage, no reload
print(RealImage.load_count)  # => loaded exactly once, despite two render() calls
# => Output: 1
# => The proxy defers construction of the expensive real subject until the FIRST access, then caches it

0
loading photo.png from disk
1

"""Example 32: pytest verification for A Virtual Proxy Defers an Expensive Load."""
 
from example import ImageProxy, RealImage
 
 
def test_constructing_the_proxy_loads_nothing() -> None:
    before: int = RealImage.load_count
    ImageProxy("unit-test.png")  # => constructing the proxy itself must stay cheap
    assert RealImage.load_count == before  # => no RealImage was constructed yet
 
 
def test_first_render_call_loads_exactly_once() -> None:
    before: int = RealImage.load_count
    proxy: ImageProxy = ImageProxy("unit-test-2.png")
    proxy.render()  # => triggers the one and only load
    proxy.render()  # => must NOT trigger a second load
    proxy.render()  # => must NOT trigger a third load
    assert RealImage.load_count == before + 1  # => loaded exactly once across 3 calls
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 33: A Protection Proxy Checks Permission Before Delegating."""
 
 
class RealBankAccount:  # => the REAL subject -- has no idea about permissions at all
    def __init__(self, balance: float) -> None:  # => the constructor
        self.balance = balance  # => stores balance on this instance
 
    def withdraw(self, amount: float) -> None:  # => defines the withdraw() method
        self.balance -= amount  # => the actual, unguarded mutation
 
 
class BankAccountProxy:  # => the PROTECTION PROXY -- adds a permission check IN FRONT
    def __init__(self, real_account: RealBankAccount, role: str) -> None:  # => holds the real subject AND the caller's role
        self._real: RealBankAccount = real_account  # => the object being protected
        self._role: str = role  # => the caller's role, checked before every delegation
 
    def withdraw(self, amount: float) -> None:  # => SAME interface as RealBankAccount
        if self._role != "admin":  # => the check RealBankAccount itself never performs
            raise PermissionError(f"role '{self._role}' cannot withdraw")  # => blocks the call entirely
        self._real.withdraw(amount)  # => only reached once the check has passed
 
 
account: RealBankAccount = RealBankAccount(balance=1000.0)  # => constructs account
admin_proxy: BankAccountProxy = BankAccountProxy(account, role="admin")  # => an authorized caller
guest_proxy: BankAccountProxy = BankAccountProxy(account, role="guest")  # => an unauthorized caller, SAME underlying account
 
admin_proxy.withdraw(100.0)  # => admin role passes the check -- delegates through
print(account.balance)  # => the withdrawal genuinely happened on the real account
# => Output: 900.0
 
try:  # => the block below is expected to raise
    guest_proxy.withdraw(50.0)  # => guest role fails the check -- never reaches RealBankAccount
except PermissionError as exc:  # => catches the PermissionError raised above
    print(exc)  # => confirms the exact rejection message
# => Output: role 'guest' cannot withdraw
print(account.balance)  # => unchanged -- the blocked withdrawal never touched the real account
# => Output: 900.0
# => The protection proxy enforces a permission check that the real subject itself does not know exists

900.0
role 'guest' cannot withdraw
900.0

"""Example 33: pytest verification for A Protection Proxy Checks Permission."""
 
import pytest
 
from example import BankAccountProxy, RealBankAccount
 
 
def test_admin_role_withdrawal_succeeds() -> None:
    account: RealBankAccount = RealBankAccount(balance=1000.0)
    BankAccountProxy(account, role="admin").withdraw(100.0)
    assert account.balance == 900.0  # => the real account was genuinely mutated
 
 
def test_guest_role_withdrawal_is_blocked() -> None:
    account: RealBankAccount = RealBankAccount(balance=1000.0)
    with pytest.raises(PermissionError):  # => must raise, never silently succeed
        BankAccountProxy(account, role="guest").withdraw(50.0)
    assert account.balance == 1000.0  # => the blocked call never reached the real account
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 34: Composite Treats File and Directory Uniformly for size()."""
 
import abc  # => imports the abc module
 
 
class FileSystemEntry(abc.ABC):  # => the COMPONENT interface -- shared by leaf AND composite
    @abc.abstractmethod
    def size(self) -> int:  # => no body -- required by both File and Directory
        ...  # => the ellipsis stub -- File and Directory below fill this in
 
 
class File(FileSystemEntry):  # => the LEAF -- has no children of its own
    def __init__(self, name: str, size_bytes: int) -> None:  # => the constructor
        self.name = name  # => stores name on this instance
        self._size_bytes = size_bytes  # => stores this leaf's own, fixed size
 
    def size(self) -> int:  # => the leaf case -- just returns its own size
        return self._size_bytes  # => returns this value to the caller
 
 
class Directory(FileSystemEntry):  # => the COMPOSITE -- holds a list of ANY FileSystemEntry
    def __init__(self, name: str) -> None:  # => the constructor
        self.name = name  # => stores name on this instance
        self._children: list[FileSystemEntry] = []  # => Files AND Directories, mixed freely
 
    def add(self, entry: FileSystemEntry) -> None:  # => defines the add() method
        self._children.append(entry)  # => accepts a File OR another Directory, identically
 
    def size(self) -> int:  # => the composite case -- sums every child, RECURSIVELY
        return sum(child.size() for child in self._children)  # => each child.size() call may itself recurse into a nested Directory
 
 
root: Directory = Directory("project")  # => constructs root
root.add(File("readme.md", 120))  # => a leaf, added directly to the root
src: Directory = Directory("src")  # => a NESTED composite
src.add(File("main.py", 300))  # => a leaf inside the nested directory
src.add(File("utils.py", 180))  # => a second leaf inside the nested directory
root.add(src)  # => the nested Directory is itself just another FileSystemEntry
print(root.size())  # => 120 + (300 + 180) -- computed through ONE uniform size() call
# => Output: 600
# => `root.size()` never checks whether a child is a File or a Directory -- both answer size() the same way

600

"""Example 34: pytest verification for Composite File Tree size()."""
 
from example import Directory, File
 
 
def test_recursive_size_sums_nested_directories_through_one_interface() -> None:
    root: Directory = Directory("project")
    root.add(File("readme.md", 120))
    src: Directory = Directory("src")
    src.add(File("main.py", 300))
    src.add(File("utils.py", 180))
    root.add(src)  # => a Directory nested inside another Directory
    assert root.size() == 600  # => 120 + 300 + 180, computed recursively
 
 
def test_a_single_file_answers_size_directly() -> None:
    assert File("a.txt", 42).size() == 42  # => the leaf case needs no recursion at all
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 35: Composite Renders Nested Menu Items Uniformly."""
 
import abc  # => imports the abc module
 
 
class MenuComponent(abc.ABC):  # => the COMPONENT interface -- shared by leaf and group
    @abc.abstractmethod
    def render(self, indent: int = 0) -> str:  # => no body -- required by both
        ...  # => the ellipsis stub -- MenuItem and Menu below fill this in
 
 
class MenuItem(MenuComponent):  # => the LEAF -- a single, non-nestable entry
    def __init__(self, label: str) -> None:  # => the constructor
        self.label = label  # => stores label on this instance
 
    def render(self, indent: int = 0) -> str:  # => the leaf case -- one indented line
        return " " * indent + f"- {self.label}"  # => returns this value to the caller
 
 
class Menu(MenuComponent):  # => the COMPOSITE -- a named GROUP of MenuComponents
    def __init__(self, label: str) -> None:  # => the constructor
        self.label = label  # => stores label on this instance
        self._items: list[MenuComponent] = []  # => MenuItems AND nested Menus, mixed freely
 
    def add(self, item: MenuComponent) -> None:  # => defines the add() method
        self._items.append(item)  # => accepts a MenuItem OR another Menu, identically
 
    def render(self, indent: int = 0) -> str:  # => the composite case -- itself PLUS children
        header: str = " " * indent + f"+ {self.label}"  # => this group's own header line
        child_lines: list[str] = [item.render(indent + 2) for item in self._items]  # => each child renders itself, MORE indented, possibly recursing again
        return "\n".join([header, *child_lines])  # => returns this value to the caller
 
 
file_menu: Menu = Menu("File")  # => constructs file_menu
file_menu.add(MenuItem("New"))  # => a leaf, added directly
file_menu.add(MenuItem("Open"))  # => a leaf, added directly
recent: Menu = Menu("Open Recent")  # => a NESTED composite submenu
recent.add(MenuItem("project.py"))  # => a leaf inside the nested submenu
file_menu.add(recent)  # => the nested Menu is itself just another MenuComponent
print(file_menu.render())  # => leaves and the nested group render through ONE call
# => Output: + File
# =>           - New
# =>           - Open
# =>           + Open Recent
# =>             - project.py
# => `file_menu.render()` never branches on whether an item is a MenuItem or a nested Menu

+ File
  - New
  - Open
  + Open Recent
    - project.py

"""Example 35: pytest verification for Composite Menu Rendering."""
 
from example import Menu, MenuItem
 
 
def test_leaf_and_nested_group_render_through_one_interface() -> None:
    file_menu: Menu = Menu("File")
    file_menu.add(MenuItem("New"))
    recent: Menu = Menu("Open Recent")
    recent.add(MenuItem("project.py"))
    file_menu.add(recent)  # => a Menu nested inside another Menu
    rendered: str = file_menu.render()
    assert "+ File" in rendered  # => the top-level group's own header
    assert "- New" in rendered  # => a direct leaf item
    assert "+ Open Recent" in rendered  # => the nested group's own header
    assert "- project.py" in rendered  # => a leaf inside the nested group
 
 
def test_nested_items_are_indented_more_than_their_parent() -> None:
    root: Menu = Menu("Root")
    root.add(MenuItem("Top"))
    lines: list[str] = root.render().splitlines()
    assert lines[0] == "+ Root"  # => no indentation at the top level
    assert lines[1] == "  - Top"  # => one nesting level deeper, two spaces indented
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 36: Command Objects with execute()/undo() for an Editor."""
 
import abc  # => imports the abc module
 
 
class Editor:  # => the RECEIVER -- the object commands actually act upon
    def __init__(self) -> None:  # => the constructor
        self.text: str = ""  # => the editor's own, mutable document content
 
    def append(self, chunk: str) -> None:  # => defines the append() method
        self.text += chunk  # => the low-level operation a Command wraps
 
    def remove_last(self, length: int) -> None:  # => defines the remove_last() method
        self.text = self.text[:-length]  # => the low-level UNDO operation
 
 
class Command(abc.ABC):  # => reifies a REQUEST as an object, with both directions
    @abc.abstractmethod
    def execute(self) -> None:  # => no body -- required by every concrete command
        ...  # => the ellipsis stub -- AppendCommand below fills this in
 
    @abc.abstractmethod
    def undo(self) -> None:  # => no body -- required by every concrete command
        ...  # => the ellipsis stub -- AppendCommand below fills this in
 
 
class AppendCommand(Command):  # => a CONCRETE command wrapping one append operation
    def __init__(self, editor: Editor, chunk: str) -> None:  # => the constructor
        self._editor: Editor = editor  # => the receiver this command acts on
        self._chunk: str = chunk  # => remembers WHAT it appended, needed to undo later
 
    def execute(self) -> None:  # => defines the execute() method
        self._editor.append(self._chunk)  # => delegates to the receiver's low-level op
 
    def undo(self) -> None:  # => defines the undo() method
        self._editor.remove_last(len(self._chunk))  # => reverses exactly what execute() did
 
 
history: list[Command] = []  # => a stack of EXECUTED commands, most recent last
editor: Editor = Editor()  # => constructs editor
 
for chunk in ["Hello", ", ", "World"]:  # => three separate edits, one command each
    cmd: Command = AppendCommand(editor, chunk)  # => wraps this edit as a Command object
    cmd.execute()  # => performs the edit
    history.append(cmd)  # => remembered so it can be undone later, in reverse order
 
print(editor.text)  # => all three chunks appended, in order
# => Output: Hello, World
 
last: Command = history.pop()  # => the MOST RECENT command, undone first
last.undo()  # => reverses only the last edit, "World"
print(editor.text)  # => "World" is gone -- the two earlier edits remain intact
# => Output: Hello,
# => Wrapping each edit as a Command object turns "undo the last thing" into `history.pop().undo()`

Hello, World
Hello,

"""Example 36: pytest verification for Command Objects with execute()/undo()."""
 
from example import AppendCommand, Command, Editor
 
 
def test_undo_reverses_only_the_most_recently_executed_command() -> None:
    editor: Editor = Editor()
    history: list[Command] = []
    for chunk in ["Hello", ", ", "World"]:
        cmd: Command = AppendCommand(editor, chunk)
        cmd.execute()
        history.append(cmd)
    assert editor.text == "Hello, World"  # => all three edits applied, in order
 
    history.pop().undo()  # => undoes only "World"
    assert editor.text == "Hello, "  # => the two earlier edits are still intact
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 37: Queueing Commands for Deferred Batch Execution."""
 
import abc  # => imports the abc module
 
 
class Command(abc.ABC):  # => reifies a REQUEST as an object, queueable like any value
    @abc.abstractmethod
    def execute(self) -> str:  # => no body -- required by every concrete command
        ...  # => the ellipsis stub -- concrete commands below fill this in
 
 
class SendEmailCommand(Command):  # => a CONCRETE command -- not run until the queue drains
    def __init__(self, to: str) -> None:  # => the constructor
        self._to: str = to  # => remembers WHAT to do, without doing it yet
 
    def execute(self) -> str:  # => defines the execute() method
        return f"email sent to {self._to}"  # => the actual side effect, deferred until now
 
 
class LogEventCommand(Command):  # => a DIFFERENT concrete command, same Command interface
    def __init__(self, message: str) -> None:  # => the constructor
        self._message: str = message  # => remembers WHAT to do, without doing it yet
 
    def execute(self) -> str:  # => defines the execute() method
        return f"logged: {self._message}"  # => the actual side effect, deferred until now
 
 
class CommandQueue:  # => holds commands as PLAIN DATA until something drains the queue
    def __init__(self) -> None:  # => the constructor
        self._pending: list[Command] = []  # => nothing executes just by being queued
 
    def enqueue(self, command: Command) -> None:  # => defines the enqueue() method
        self._pending.append(command)  # => appends WITHOUT calling execute() at all
 
    def run_all(self) -> list[str]:  # => the deferred batch execution step
        results: list[str] = [cmd.execute() for cmd in self._pending]  # => runs every queued command, in the EXACT order they were enqueued
        self._pending.clear()  # => the queue is empty once this batch has run
        return results  # => returns this value to the caller
 
 
queue: CommandQueue = CommandQueue()  # => constructs queue
queue.enqueue(LogEventCommand("user signed up"))  # => queued FIRST -- not yet executed
queue.enqueue(SendEmailCommand("new-user@example.com"))  # => queued SECOND -- not yet executed
queue.enqueue(LogEventCommand("welcome email queued"))  # => queued THIRD -- not yet executed
 
print(len(queue._pending))  # => nothing has executed yet -- just three unexecuted commands, queued
# => Output: 3
results: list[str] = queue.run_all()  # => the ENTIRE batch runs now, in enqueue order
for line in results:  # => prints each result on its own line
    print(line)  # => confirms execution happened in the SAME order commands were enqueued
# => Output: logged: user signed up
# => Output: email sent to new-user@example.com
# => Output: logged: welcome email queued
# => Queueing a Command defers its side effect; `run_all()` executes every queued command in enqueue order

3
logged: user signed up
email sent to new-user@example.com
logged: welcome email queued

"""Example 37: pytest verification for Queueing Commands for Deferred Batch Execution."""
 
from example import CommandQueue, LogEventCommand, SendEmailCommand
 
 
def test_run_all_executes_commands_in_enqueue_order() -> None:
    queue: CommandQueue = CommandQueue()
    queue.enqueue(LogEventCommand("first"))
    queue.enqueue(SendEmailCommand("second@example.com"))
    queue.enqueue(LogEventCommand("third"))
    results: list[str] = queue.run_all()
    assert results == [
        "logged: first",
        "email sent to second@example.com",
        "logged: third",
    ]  # => exact enqueue order preserved
 
 
def test_run_all_empties_the_queue() -> None:
    queue: CommandQueue = CommandQueue()
    queue.enqueue(LogEventCommand("only"))
    queue.run_all()
    assert queue._pending == []  # => nothing left pending after a batch has run
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 38: A Vending Machine State Machine, Not Boolean Flags."""
 
import abc  # => imports the abc module
 
 
class VendingMachine:  # => the CONTEXT -- holds whichever state object is CURRENT
    def __init__(self) -> None:  # => the constructor
        self.state: "MachineState" = NoCoinState()  # => starts in the NoCoin state
 
    def insert_coin(self) -> str:  # => delegates the DECISION to the current state object
        return self.state.insert_coin(self)  # => returns this value to the caller
 
    def dispense(self) -> str:  # => delegates the DECISION to the current state object
        return self.state.dispense(self)  # => returns this value to the caller
 
 
class MachineState(abc.ABC):  # => one state object PER machine state, not a flag/enum
    @abc.abstractmethod
    def insert_coin(self, machine: VendingMachine) -> str:  # => no body -- required
        ...  # => the ellipsis stub -- concrete states below fill this in
 
    @abc.abstractmethod
    def dispense(self, machine: VendingMachine) -> str:  # => no body -- required
        ...  # => the ellipsis stub -- concrete states below fill this in
 
 
class NoCoinState(MachineState):  # => the machine has NO coin inserted yet
    def insert_coin(self, machine: VendingMachine) -> str:  # => the LEGAL transition here
        machine.state = HasCoinState()  # => switches the context to a DIFFERENT state object
        return "coin accepted"  # => returns this value to the caller
 
    def dispense(self, machine: VendingMachine) -> str:  # => the ILLEGAL transition here
        raise ValueError("insert a coin first")  # => rejected -- no coin, nothing to dispense
 
 
class HasCoinState(MachineState):  # => the machine already HAS a coin inserted
    def insert_coin(self, machine: VendingMachine) -> str:  # => the ILLEGAL transition here
        raise ValueError("coin already inserted")  # => rejected -- can't insert a second coin
 
    def dispense(self, machine: VendingMachine) -> str:  # => the LEGAL transition here
        machine.state = NoCoinState()  # => switches back to the NoCoin state
        return "item dispensed"  # => returns this value to the caller
 
 
machine: VendingMachine = VendingMachine()  # => constructs machine
print(machine.insert_coin())  # => legal from NoCoinState
# => Output: coin accepted
print(machine.dispense())  # => legal from HasCoinState -- item is released
# => Output: item dispensed
 
try:  # => the block below is expected to raise
    machine.dispense()  # => ILLEGAL: back in NoCoinState, nothing to dispense
except ValueError as exc:  # => catches the ValueError raised above
    print(exc)  # => confirms the illegal transition was rejected, not silently ignored
# => Output: insert a coin first
# => Each state is its OWN object that decides which transitions are legal, replacing scattered if/flag checks

coin accepted
item dispensed
insert a coin first

"""Example 38: pytest verification for the Vending Machine State Machine."""
 
import pytest
 
from example import VendingMachine
 
 
def test_legal_transitions_move_through_both_states() -> None:
    machine: VendingMachine = VendingMachine()
    assert machine.insert_coin() == "coin accepted"  # => NoCoin -> HasCoin
    assert machine.dispense() == "item dispensed"  # => HasCoin -> NoCoin
 
 
def test_illegal_dispense_without_a_coin_is_rejected() -> None:
    machine: VendingMachine = VendingMachine()
    with pytest.raises(ValueError):  # => dispensing with no coin must raise, not succeed
        machine.dispense()
 
 
def test_illegal_double_coin_insert_is_rejected() -> None:
    machine: VendingMachine = VendingMachine()
    machine.insert_coin()  # => now in HasCoinState
    with pytest.raises(ValueError):  # => a second coin insert must raise, not succeed
        machine.insert_coin()
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 39: Cycling States via the State Pattern."""
 
import abc  # => imports the abc module
 
 
class TrafficLight:  # => the CONTEXT -- holds whichever LightState object is CURRENT
    def __init__(self) -> None:  # => the constructor
        self.state: "LightState" = RedState()  # => every light starts at Red
 
    def next(self) -> str:  # => delegates "what comes next" to the CURRENT state object
        self.state = self.state.next_state()  # => the current state DECIDES its successor
        return self.state.name()  # => returns this value to the caller
 
 
class LightState(abc.ABC):  # => one state object PER color, not an enum with a switch
    @abc.abstractmethod
    def next_state(self) -> "LightState":  # => no body -- required by every color
        ...  # => the ellipsis stub -- concrete states below fill this in
 
    @abc.abstractmethod
    def name(self) -> str:  # => no body -- required by every color
        ...  # => the ellipsis stub -- concrete states below fill this in
 
 
class RedState(LightState):  # => Red always transitions to Green next
    def next_state(self) -> LightState:  # => defines the next_state() method
        return GreenState()  # => Red -> Green, and ONLY Red -> Green
 
    def name(self) -> str:  # => defines the name() method
        return "red"  # => returns this value to the caller
 
 
class GreenState(LightState):  # => Green always transitions to Yellow next
    def next_state(self) -> LightState:  # => defines the next_state() method
        return YellowState()  # => Green -> Yellow, and ONLY Green -> Yellow
 
    def name(self) -> str:  # => defines the name() method
        return "green"  # => returns this value to the caller
 
 
class YellowState(LightState):  # => Yellow always transitions back to Red
    def next_state(self) -> LightState:  # => defines the next_state() method
        return RedState()  # => Yellow -> Red, closing the cycle
 
    def name(self) -> str:  # => defines the name() method
        return "yellow"  # => returns this value to the caller
 
 
light: TrafficLight = TrafficLight()  # => constructs light, starting at Red
print(light.state.name())  # => the initial state, before any next() call
# => Output: red
print(light.next())  # => Red -> Green
# => Output: green
print(light.next())  # => Green -> Yellow
# => Output: yellow
print(light.next())  # => Yellow -> Red, the cycle repeats
# => Output: red
# => Each LightState object owns its OWN transition rule -- `next()` never branches on the color by name

red
green
yellow
red

"""Example 39: pytest verification for Cycling States via the State Pattern."""
 
from example import TrafficLight
 
 
def test_next_cycles_red_green_yellow_red() -> None:
    light: TrafficLight = TrafficLight()
    assert light.state.name() == "red"  # => starting state
    assert light.next() == "green"  # => red -> green
    assert light.next() == "yellow"  # => green -> yellow
    assert light.next() == "red"  # => yellow -> red, cycle closes
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 40: A Custom __iter__ Walks a Binary Tree In Order."""
 
from collections.abc import Iterator  # => imports Iterator from collections.abc
 
 
class Node:  # => a single binary tree node -- NOT itself iterable
    def __init__(self, value: int, left: "Node | None" = None, right: "Node | None" = None) -> None:  # => the constructor
        self.value = value  # => stores value on this instance
        self.left = left  # => stores left on this instance
        self.right = right  # => stores right on this instance
 
 
class BinaryTree:  # => the CONTAINER -- exposes iteration without exposing Node internals
    def __init__(self, root: Node | None) -> None:  # => the constructor
        self.root = root  # => stores root on this instance
 
    def __iter__(self) -> Iterator[int]:  # => makes `for v in tree` work directly
        yield from self._in_order(self.root)  # => delegates to a recursive generator helper
 
    def _in_order(self, node: Node | None) -> Iterator[int]:  # => LEFT, self, RIGHT
        if node is None:  # => the recursion's base case -- nothing to yield
            return  # => stops this generator branch cleanly
        yield from self._in_order(node.left)  # => everything smaller, first
        yield node.value  # => this node's own value, in the middle
        yield from self._in_order(node.right)  # => everything larger, last
 
 
#        4
#       / \
#      2   6
#     / \   \
#    1   3   7
tree: BinaryTree = BinaryTree(Node(4, Node(2, Node(1), Node(3)), Node(6, None, Node(7))))  # => a small, deliberately unbalanced tree
values: list[int] = [v for v in tree]  # => a plain for-loop -- no BinaryTree internals exposed
print(values)  # => in-order traversal yields values in SORTED order for a binary search tree
# => Output: [1, 2, 3, 4, 6, 7]
# => `__iter__` returning a generator lets `for v in tree` walk arbitrarily deep structure lazily

[1, 2, 3, 4, 6, 7]

"""Example 40: pytest verification for the Custom __iter__ Binary Tree Walk."""
 
from example import BinaryTree, Node
 
 
def test_for_loop_yields_values_in_sorted_order() -> None:
    tree: BinaryTree = BinaryTree(Node(4, Node(2, Node(1), Node(3)), Node(6, None, Node(7))))
    assert [v for v in tree] == [1, 2, 3, 4, 6, 7]  # => in-order == sorted, for a BST
 
 
def test_empty_tree_yields_nothing() -> None:
    tree: BinaryTree = BinaryTree(None)
    assert list(tree) == []  # => no root -- the generator yields zero values
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 41: An Iterator That Lazily Pages a Remote API."""
 
from collections.abc import Iterator  # => imports Iterator from collections.abc
 
# => a self-contained FAKE standing in for a real, network-backed API -- no real network call
_FAKE_REMOTE_PAGES: dict[int, list[str]] = {
    1: ["item-a", "item-b"],  # => page 1's two records
    2: ["item-c", "item-d"],  # => page 2's two records
    3: [],  # => an EMPTY page signals "no more data" -- the fetch loop stops here
}
 
 
class FetchLog:  # => an observable stand-in for "a network call happened"
    def __init__(self) -> None:  # => the constructor
        self.fetched_pages: list[int] = []  # => records EVERY page number actually fetched
 
 
def fetch_page(page_number: int, log: FetchLog) -> list[str]:  # => the "remote" call
    log.fetched_pages.append(page_number)  # => records that THIS page was fetched now
    return list(_FAKE_REMOTE_PAGES.get(page_number, []))  # => a COPY -- callers pop() their own buffer, never the shared fake dataset
 
 
class PagedApiIterator:  # => fetches ONE page at a time, only when the current page runs out
    def __init__(self, log: FetchLog) -> None:  # => the constructor
        self._log: FetchLog = log  # => shared so the example can OBSERVE fetch timing
        self._page_number: int = 1  # => starts at the first page
        self._buffer: list[str] = []  # => the current page's not-yet-yielded items
        self._exhausted: bool = False  # => True once an empty page has been seen
 
    def __iter__(self) -> Iterator[str]:  # => makes `for item in paged_iterator` work
        while True:  # => keeps pulling pages until the fake API returns an empty one
            if not self._buffer and not self._exhausted:  # => current page is used up
                self._buffer = fetch_page(self._page_number, self._log)  # => fetches lazily, exactly when needed
                self._page_number += 1  # => advances for the NEXT fetch, if one happens
                if not self._buffer:  # => an empty page means there is nothing left
                    self._exhausted = True  # => stops future fetch attempts
            if not self._buffer:  # => truly nothing left to yield
                return  # => stops the generator cleanly
            yield self._buffer.pop(0)  # => yields ONE item at a time, oldest first
 
 
log: FetchLog = FetchLog()  # => constructs log
paged: PagedApiIterator = PagedApiIterator(log)  # => cheap -- no page has been fetched yet
print(log.fetched_pages)  # => confirms NOTHING was fetched just by constructing the iterator
# => Output: []
 
first_item: str = next(iter(paged))  # => pulling ONE item triggers fetching page 1 only
print(first_item, log.fetched_pages)  # => page 2 was NOT fetched just to get one item
# => Output: item-a [1]
 
all_items: list[str] = list(paged)  # => draining the rest pulls page 1's remainder + page 2
print(all_items)  # => item-b was still buffered; item-c/item-d came from the page-2 fetch
# => Output: ['item-b', 'item-c', 'item-d']
print(log.fetched_pages)  # => page 3 (empty) was fetched too, to discover the data had ended
# => Output: [1, 2, 3]
# => Pages are fetched lazily, one at a time, only when the current buffer is exhausted

[]
item-a [1]
['item-b', 'item-c', 'item-d']
[1, 2, 3]

"""Example 41: pytest verification for the Lazily-Paged API Iterator."""
 
from example import FetchLog, PagedApiIterator
 
 
def test_constructing_the_iterator_fetches_nothing() -> None:
    log: FetchLog = FetchLog()
    PagedApiIterator(log)  # => construction alone must not touch the fake API
    assert log.fetched_pages == []
 
 
def test_pages_are_fetched_lazily_one_at_a_time() -> None:
    log: FetchLog = FetchLog()
    paged: PagedApiIterator = PagedApiIterator(log)
    first: str = next(iter(paged))  # => only enough to produce one item
    assert first == "item-a"
    assert log.fetched_pages == [1]  # => page 2 was NOT fetched yet
 
 
def test_draining_the_iterator_fetches_every_remaining_page() -> None:
    log: FetchLog = FetchLog()
    paged: PagedApiIterator = PagedApiIterator(log)
    all_items: list[str] = list(paged)  # => drains everything, across all pages
    assert all_items == ["item-a", "item-b", "item-c", "item-d"]
    assert log.fetched_pages == [1, 2, 3]  # => including the empty terminating page
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 42: Escalating a Support Ticket Through a Handler Chain."""
 
import abc  # => imports the abc module
 
 
class SupportHandler(abc.ABC):  # => one link in the chain -- each link decides FOR ITSELF
    def __init__(self) -> None:  # => the constructor
        self._next: "SupportHandler | None" = None  # => the NEXT link, set by set_next()
 
    def set_next(self, handler: "SupportHandler") -> "SupportHandler":  # => wires the chain
        self._next = handler  # => remembers who to escalate to
        return handler  # => returned so calls can be chained: a.set_next(b).set_next(c)
 
    def handle(self, severity: int) -> str:  # => defines the handle() method
        if self._can_handle(severity):  # => THIS link decides whether it owns the ticket
            return self._resolve(severity)  # => returns this value to the caller
        if self._next is not None:  # => not this link's job -- pass it further down the chain
            return self._next.handle(severity)  # => returns this value to the caller
        return "unhandled: no tier could resolve this ticket"  # => fell off the END of the chain
 
    @abc.abstractmethod
    def _can_handle(self, severity: int) -> bool:  # => no body -- required by every tier
        ...  # => the ellipsis stub -- concrete tiers below fill this in
 
    @abc.abstractmethod
    def _resolve(self, severity: int) -> str:  # => no body -- required by every tier
        ...  # => the ellipsis stub -- concrete tiers below fill this in
 
 
class L1Handler(SupportHandler):  # => handles only the LOWEST severities
    def _can_handle(self, severity: int) -> bool:  # => defines the _can_handle() method
        return severity <= 1  # => returns this value to the caller
 
    def _resolve(self, severity: int) -> str:  # => defines the _resolve() method
        return "resolved at L1"  # => returns this value to the caller
 
 
class L2Handler(SupportHandler):  # => handles the NEXT band of severities
    def _can_handle(self, severity: int) -> bool:  # => defines the _can_handle() method
        return severity <= 3  # => returns this value to the caller
 
    def _resolve(self, severity: int) -> str:  # => defines the _resolve() method
        return "resolved at L2"  # => returns this value to the caller
 
 
class L3Handler(SupportHandler):  # => the LAST link -- handles everything remaining
    def _can_handle(self, severity: int) -> bool:  # => defines the _can_handle() method
        return severity <= 5  # => returns this value to the caller
 
    def _resolve(self, severity: int) -> str:  # => defines the _resolve() method
        return "resolved at L3"  # => returns this value to the caller
 
 
l1: L1Handler = L1Handler()  # => constructs l1
l2: L2Handler = L2Handler()  # => constructs l2
l3: L3Handler = L3Handler()  # => constructs l3
l1.set_next(l2).set_next(l3)  # => wires L1 -> L2 -> L3, chained in one expression
 
print(l1.handle(1))  # => severity 1 -- L1 can handle it directly, no escalation needed
# => Output: resolved at L1
print(l1.handle(2))  # => severity 2 -- too high for L1, ESCALATES to L2 automatically
# => Output: resolved at L2
print(l1.handle(9))  # => severity 9 -- too high for every tier, falls off the end
# => Output: unhandled: no tier could resolve this ticket
# => An unhandled ticket automatically escalates to the NEXT handler, with no caller-side branching

resolved at L1
resolved at L2
unhandled: no tier could resolve this ticket

"""Example 42: pytest verification for Escalating a Ticket Through a Handler Chain."""
 
from example import L1Handler, L2Handler, L3Handler
 
 
def _build_chain() -> L1Handler:
    l1, l2, l3 = L1Handler(), L2Handler(), L3Handler()
    l1.set_next(l2).set_next(l3)
    return l1
 
 
def test_ticket_handled_at_the_first_capable_tier() -> None:
    assert _build_chain().handle(1) == "resolved at L1"  # => no escalation needed
 
 
def test_unhandled_ticket_at_l1_falls_to_the_next_handler() -> None:
    assert _build_chain().handle(2) == "resolved at L2"  # => L1 rejects it, L2 resolves it
 
 
def test_ticket_beyond_every_tier_reports_unhandled() -> None:
    assert _build_chain().handle(9) == "unhandled: no tier could resolve this ticket"  # => fell off the end of the chain
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 43: A Validation Pipeline as a Handler Chain."""
 
import abc  # => imports the abc module
 
 
class ValidationHandler(abc.ABC):  # => one validation RULE, chained to the next rule
    def __init__(self) -> None:  # => the constructor
        self._next: "ValidationHandler | None" = None  # => the NEXT rule in the pipeline
 
    def set_next(self, handler: "ValidationHandler") -> "ValidationHandler":  # => wires the chain
        self._next = handler  # => remembers who runs after this rule
        return handler  # => returned so calls can be chained: a.set_next(b).set_next(c)
 
    def validate(self, value: str) -> str | None:  # => None means "passed every rule"
        error: str | None = self._check(value)  # => THIS rule's own check, run first
        if error is not None:  # => the FIRST failure stops the entire chain immediately
            return error  # => returns this value to the caller
        if self._next is not None:  # => this rule passed -- let the NEXT rule run
            return self._next.validate(value)  # => returns this value to the caller
        return None  # => every rule in the chain passed
 
    @abc.abstractmethod
    def _check(self, value: str) -> str | None:  # => no body -- required by every rule
        ...  # => the ellipsis stub -- concrete rules below fill this in
 
 
class NotEmptyRule(ValidationHandler):  # => runs FIRST in the pipeline
    def _check(self, value: str) -> str | None:  # => defines the _check() method
        return "value must not be empty" if value == "" else None  # => returns this value
 
 
class MinLengthRule(ValidationHandler):  # => runs SECOND, only if NotEmptyRule passed
    def _check(self, value: str) -> str | None:  # => defines the _check() method
        return "value must be at least 4 characters" if len(value) < 4 else None  # => returns this value to the caller
 
 
class AlphaOnlyRule(ValidationHandler):  # => runs THIRD, only if the first two passed
    def _check(self, value: str) -> str | None:  # => defines the _check() method
        return "value must be alphabetic" if not value.isalpha() else None  # => returns this
 
 
not_empty: NotEmptyRule = NotEmptyRule()  # => constructs not_empty
min_length: MinLengthRule = MinLengthRule()  # => constructs min_length
alpha_only: AlphaOnlyRule = AlphaOnlyRule()  # => constructs alpha_only
not_empty.set_next(min_length).set_next(alpha_only)  # => wires the pipeline in ONE expression
 
print(not_empty.validate(""))  # => fails the VERY FIRST rule -- the chain stops here
# => Output: value must not be empty
print(not_empty.validate("ab"))  # => passes rule 1, fails rule 2 -- rule 3 never runs
# => Output: value must be at least 4 characters
print(not_empty.validate("abcd"))  # => passes every rule in the pipeline
# => Output: None
# => The FIRST failing rule stops the chain immediately -- later rules never even run on bad input

value must not be empty
value must be at least 4 characters
None

"""Example 43: pytest verification for the Validation Pipeline as a Handler Chain."""
 
from example import AlphaOnlyRule, MinLengthRule, NotEmptyRule
 
 
def _build_pipeline() -> NotEmptyRule:
    not_empty, min_length, alpha_only = NotEmptyRule(), MinLengthRule(), AlphaOnlyRule()
    not_empty.set_next(min_length).set_next(alpha_only)
    return not_empty
 
 
def test_first_failure_stops_the_chain_and_reports_that_error() -> None:
    assert _build_pipeline().validate("") == "value must not be empty"  # => rule 1 fails first
 
 
def test_second_rule_only_runs_once_the_first_rule_passes() -> None:
    assert _build_pipeline().validate("ab") == "value must be at least 4 characters"  # => rule 1 passed, rule 2 caught it
 
 
def test_value_passing_every_rule_returns_none() -> None:
    assert _build_pipeline().validate("abcd") is None  # => every rule in the chain passed
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 44: A Typed Event Bus with Unsubscribe."""
 
from collections.abc import Callable  # => imports Callable from collections.abc
 
Handler = Callable[[str], None]  # => a type alias: any function taking str, returning None
 
 
class EventBus:  # => the SUBJECT -- notifies subscribers without knowing their concrete type
    def __init__(self) -> None:  # => the constructor
        self._handlers: dict[str, list[Handler]] = {}  # => topic name -> list of subscribers
 
    def subscribe(self, topic: str, handler: Handler) -> None:  # => defines subscribe()
        self._handlers.setdefault(topic, []).append(handler)  # => adds this handler to the topic's list
 
    def unsubscribe(self, topic: str, handler: Handler) -> None:  # => defines unsubscribe()
        self._handlers.get(topic, []).remove(handler)  # => removes ONLY this handler, leaves others intact
 
    def publish(self, topic: str, message: str) -> None:  # => defines the publish() method
        for handler in self._handlers.get(topic, []):  # => notifies EVERY currently-subscribed handler
            handler(message)  # => the bus never knows or cares what each handler DOES
 
 
received: list[str] = []  # => a plain list used to OBSERVE which handlers actually fired
 
 
def log_handler(message: str) -> None:  # => one concrete subscriber
    received.append(f"log: {message}")  # => records that this handler ran
 
 
def alert_handler(message: str) -> None:  # => a DIFFERENT concrete subscriber
    received.append(f"alert: {message}")  # => records that this handler ran
 
 
bus: EventBus = EventBus()  # => constructs bus
bus.subscribe("order.created", log_handler)  # => subscribes log_handler to this topic
bus.subscribe("order.created", alert_handler)  # => subscribes alert_handler to the SAME topic
 
bus.publish("order.created", "order-1")  # => notifies BOTH currently-subscribed handlers
print(received)  # => both handlers ran, in subscription order
# => Output: ['log: order-1', 'alert: order-1']
 
bus.unsubscribe("order.created", alert_handler)  # => alert_handler is no longer subscribed
received.clear()  # => resets the observation list for the next publish
bus.publish("order.created", "order-2")  # => notifies only whichever handlers remain
print(received)  # => alert_handler did NOT run -- it was unsubscribed above
# => Output: ['log: order-2']
# => `unsubscribe` removes exactly one handler from one topic; every other subscription is unaffected

['log: order-1', 'alert: order-1']
['log: order-2']

"""Example 44: pytest verification for the Typed Event Bus with Unsubscribe."""
 
from example import EventBus
 
 
def test_every_subscribed_handler_receives_a_published_event() -> None:
    received: list[str] = []
    bus: EventBus = EventBus()
    bus.subscribe("topic", lambda msg: received.append(f"a:{msg}"))
    bus.subscribe("topic", lambda msg: received.append(f"b:{msg}"))
    bus.publish("topic", "hello")
    assert received == ["a:hello", "b:hello"]  # => both handlers fired
 
 
def test_unsubscribed_handler_is_not_called() -> None:
    received: list[str] = []
 
    def handler(msg: str) -> None:
        received.append(msg)
 
    bus: EventBus = EventBus()
    bus.subscribe("topic", handler)
    bus.unsubscribe("topic", handler)  # => removed BEFORE any publish() call
    bus.publish("topic", "hello")
    assert received == []  # => the unsubscribed handler never ran
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 45: A Strategy via typing.Protocol Duck Typing."""
 
from typing import Protocol  # => imports Protocol from typing
 
 
class PricingStrategy(Protocol):  # => a STRUCTURAL type -- "anything callable like this"
    def __call__(self, base: float) -> float:  # => the shape every strategy must match
        ...  # => the ellipsis stub -- no shared base class required anywhere
 
 
def regular_pricing(base: float) -> float:  # => a PLAIN FUNCTION -- never declares Protocol
    return base  # => returns this value to the caller
 
 
def discount_pricing(base: float) -> float:  # => a DIFFERENT plain function, same shape
    return base * 0.8  # => returns this value to the caller
 
 
class Order:  # => begins the Order class body
    def __init__(self, pricing: PricingStrategy) -> None:  # => type-hinted against the PROTOCOL, not a concrete class
        self.pricing = pricing  # => held as a swappable collaborator
 
    def total(self, base: float) -> float:  # => defines the total() method
        return self.pricing(base)  # => calls the strategy as an ordinary function
 
 
regular_order: Order = Order(regular_pricing)  # => a bare FUNCTION satisfies the Protocol
discount_order: Order = Order(discount_pricing)  # => a DIFFERENT bare function, same Protocol
print(regular_order.total(100.0))  # => regular_pricing applied
# => Output: 100.0
print(discount_order.total(100.0))  # => discount_pricing applied -- same Order class
# => Output: 80.0
# => `regular_pricing` never declares `PricingStrategy` anywhere -- structural typing accepts it by shape alone

100.0
80.0

"""Example 45: pytest verification for the Strategy via typing.Protocol."""
 
from example import Order, discount_pricing, regular_pricing
 
 
def test_plain_function_satisfies_the_protocol_and_prices_regularly() -> None:
    order: Order = Order(regular_pricing)  # => a bare function, no inheritance declared
    assert order.total(100.0) == 100.0
 
 
def test_a_different_plain_function_satisfies_the_same_protocol() -> None:
    order: Order = Order(discount_pricing)  # => a DIFFERENT bare function, same shape
    assert order.total(100.0) == 80.0
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 46: Replacing a Type-Switch with Polymorphic Dispatch."""  # => module docstring
 
import abc  # => imports the abc module
 
 
class ShapeData:  # => a plain data holder, no behavior -- used only by the BEFORE version
    def __init__(self, kind: str, side: float) -> None:  # => the constructor
        self.kind = kind  # => stores kind on this instance
        self.side = side  # => stores side on this instance
 
 
def area_by_type_switch(shape: ShapeData) -> float:  # => the BEFORE version -- a type-switch
    if shape.kind == "square":  # => branch 1 -- must be edited for every NEW shape kind
        return shape.side * shape.side  # => returns this value to the caller
    if shape.kind == "circle":  # => branch 2 -- must be edited for every NEW shape kind
        return 3.14159 * shape.side * shape.side  # => returns this value to the caller
    raise ValueError(f"unknown kind: {shape.kind}")  # => a THIRD kind requires editing this exact function
 
 
class Shape(abc.ABC):  # => the AFTER version -- one abstract method, no switch anywhere
    @abc.abstractmethod  # => marks the next method as required for every Shape subclass
    def area(self) -> float:  # => no body -- required by every concrete shape
        ...  # => the ellipsis stub -- concrete shapes below fill this in
 
 
class Square(Shape):  # => the type-switch's "square" branch, now its OWN class
    def __init__(self, side: float) -> None:  # => the constructor
        self.side = side  # => stores side on this instance
 
    def area(self) -> float:  # => defines the area() method
        return self.side * self.side  # => returns this value to the caller
 
 
class Circle(Shape):  # => the type-switch's "circle" branch, now its OWN class
    def __init__(self, radius: float) -> None:  # => the constructor
        self.radius = radius  # => stores radius on this instance
 
    def area(self) -> float:  # => defines the area() method
        return 3.14159 * self.radius * self.radius  # => returns this value to the caller
 
 
class Triangle(Shape):  # => a BRAND NEW shape -- adding this edits NOTHING above it
    def __init__(self, base: float, height: float) -> None:  # => the constructor
        self.base = base  # => stores base on this instance
        self.height = height  # => stores height on this instance
 
    def area(self) -> float:  # => defines the area() method
        return 0.5 * self.base * self.height  # => returns this value to the caller
 
 
print(area_by_type_switch(ShapeData("square", 4.0)))  # => the OLD, switch-based version
# => Output: 16.0
 
shapes: list[Shape] = [  # => a heterogeneous list -- every element satisfies the Shape contract
    Square(4.0),  # => side length 4.0, same abstract type as its siblings below
    Circle(2.0),  # => radius 2.0, same abstract type as its siblings
    Triangle(3.0, 6.0),  # => base 3.0, height 6.0 -- a brand-new shape class
]  # => Triangle added with ZERO edits to any existing dispatch code
areas: list[float] = [s.area() for s in shapes]  # => ONE call-site, dispatches polymorphically for ALL three
print(areas)  # => each shape computed its OWN area, no isinstance() or kind check anywhere
# => Output: [16.0, 12.56636, 9.0]
# => Adding `Triangle` required a new class, but edited zero lines of the polymorphic dispatch loop

16.0
[16.0, 12.56636, 9.0]

"""Example 46: pytest verification for Replacing a Type-Switch with Polymorphism."""
 
from example import Circle, Square, Triangle, area_by_type_switch, ShapeData
 
 
def test_type_switch_still_computes_the_correct_area() -> None:
    assert area_by_type_switch(ShapeData("square", 4.0)) == 16.0
 
 
def test_polymorphic_dispatch_computes_every_shapes_own_area() -> None:
    shapes = [Square(4.0), Circle(2.0), Triangle(3.0, 6.0)]
    areas: list[float] = [s.area() for s in shapes]  # => one shared call-site
    assert areas == [16.0, 12.56636, 9.0]
 
 
def test_adding_a_new_shape_requires_no_dispatch_code_changes() -> None:
    # => Triangle is used here exactly like Square/Circle -- proof the dispatch loop above
    # => never needed editing when this class was added to the module
    assert Triangle(2.0, 5.0).area() == 5.0
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 47: A Repository -- a Pure Fabrication for Persistence."""
 
from dataclasses import dataclass  # => imports dataclass from dataclasses
 
 
@dataclass(frozen=True)  # => a genuine DOMAIN concept -- knows nothing about storage
class User:  # => begins the User class body
    user_id: int  # => a required field, part of the generated __init__
    email: str  # => a required field, part of the generated __init__
 
    def __post_init__(self) -> None:  # => runs automatically right after construction
        if "@" not in self.email:  # => a DOMAIN rule -- belongs on the domain object itself
            raise ValueError("email must contain @")  # => rejects an invalid User at construction time
 
 
class UserRepository:  # => a PURE FABRICATION -- no analyst would ever call this a "domain concept"
    def __init__(self) -> None:  # => invented purely to keep User cohesive and decoupled from IO
        self._storage: dict[int, User] = {}  # => the ONLY place storage details live
 
    def save(self, user: User) -> None:  # => defines the save() method
        self._storage[user.user_id] = user  # => the IO-flavored operation, isolated HERE
 
    def find_by_id(self, user_id: int) -> User | None:  # => defines the find_by_id() method
        return self._storage.get(user_id)  # => returns this value to the caller
 
 
repo: UserRepository = UserRepository()  # => constructs repo
repo.save(User(user_id=1, email="ada@example.com"))  # => persistence lives ONLY in the repo
 
found: User | None = repo.find_by_id(1)  # => User itself never touched a dict, a file, or a DB
print(found)  # => the domain object, retrieved via the fabricated persistence seam
# => Output: User(user_id=1, email='ada@example.com')
 
missing: User | None = repo.find_by_id(99)  # => an id that was never saved
print(missing)  # => confirms a clean, storage-flavored "not found" result
# => Output: None
# => `UserRepository` exists ONLY to keep persistence out of `User` -- it is invented, not discovered in the domain

User(user_id=1, email='ada@example.com')
None

"""Example 47: pytest verification for the Repository Pure Fabrication."""
 
import inspect
 
from example import User, UserRepository
 
 
def test_user_class_defines_no_io_touching_methods() -> None:
    # => the domain class exposes only __init__/__post_init__ plus dataclass-generated
    # => dunder methods -- no save/load/find method exists on User itself
    method_names: set[str] = {name for name, _ in inspect.getmembers(User, predicate=inspect.isfunction)}
    assert "save" not in method_names  # => persistence lives on the repository, not here
    assert "find_by_id" not in method_names  # => persistence lives on the repository, not here
 
 
def test_repository_saves_and_retrieves_the_domain_object() -> None:
    repo: UserRepository = UserRepository()
    repo.save(User(user_id=1, email="ada@example.com"))
    found: User | None = repo.find_by_id(1)
    assert found == User(user_id=1, email="ada@example.com")
 
 
def test_repository_returns_none_for_an_unsaved_id() -> None:
    repo: UserRepository = UserRepository()
    assert repo.find_by_id(99) is None
 
 
# => Run: pytest -- Output: 3 passed

3 passed

"""Example 48: A Mediator Decouples Two Collaborators."""
 
 
class ChatRoom:  # => the MEDIATOR -- the ONLY object that knows about every Participant
    def __init__(self) -> None:  # => the constructor
        self._members: dict[str, "Participant"] = {}  # => name -> Participant, held HERE
 
    def register(self, participant: "Participant") -> None:  # => defines register()
        self._members[participant.name] = participant  # => the mediator learns about them
 
    def relay(self, sender: str, recipient: str, message: str) -> None:  # => defines relay()
        self._members[recipient].receive(sender, message)  # => the mediator looks up and delivers -- senders never do this themselves
 
 
class Participant:  # => a COLLABORATOR -- never holds a reference to another Participant
    def __init__(self, name: str, room: ChatRoom) -> None:  # => the constructor
        self.name = name  # => stores name on this instance
        self._room: ChatRoom = room  # => the ONLY collaborator this class ever references
        self.inbox: list[str] = []  # => messages this participant has received
 
    def send(self, recipient: str, message: str) -> None:  # => defines the send() method
        self._room.relay(self.name, recipient, message)  # => goes THROUGH the mediator, never directly to the recipient
 
    def receive(self, sender: str, message: str) -> None:  # => defines the receive() method
        self.inbox.append(f"{sender}: {message}")  # => records the delivered message
 
 
room: ChatRoom = ChatRoom()  # => constructs room
alice: Participant = Participant("alice", room)  # => alice holds ONLY a reference to room, never to bob
bob: Participant = Participant("bob", room)  # => bob holds ONLY a reference to room, never to alice
room.register(alice)  # => the mediator learns alice exists
room.register(bob)  # => the mediator learns bob exists
 
alice.send("bob", "hello")  # => routed THROUGH the mediator, not a direct alice -> bob call
print(bob.inbox)  # => the mediator delivered the message despite alice never touching bob
# => Output: ['alice: hello']
print(hasattr(alice, "bob") or "bob" in vars(alice))  # => alice holds no reference to bob at all
# => Output: False
# => Neither Participant ever references the other directly -- every interaction is routed through ChatRoom

['alice: hello']
False

"""Example 48: pytest verification for the Mediator Decoupling Two Collaborators."""
 
from example import ChatRoom, Participant
 
 
def test_message_is_delivered_through_the_mediator() -> None:
    room: ChatRoom = ChatRoom()
    alice: Participant = Participant("alice", room)
    bob: Participant = Participant("bob", room)
    room.register(alice)
    room.register(bob)
    alice.send("bob", "hello")
    assert bob.inbox == ["alice: hello"]  # => delivered despite no direct reference
 
 
def test_neither_participant_holds_a_reference_to_the_other() -> None:
    room: ChatRoom = ChatRoom()
    alice: Participant = Participant("alice", room)
    bob: Participant = Participant("bob", room)
    assert "bob" not in vars(alice).values()  # => alice's own attributes never name bob
    assert "alice" not in vars(bob).values()  # => bob's own attributes never name alice
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 49: Stabilizing an Unstable Vendor API Behind an Interface."""
 
import abc  # => imports the abc module
 
 
class PaymentGateway(abc.ABC):  # => the STABLE interface -- protects Checkout from vendor churn
    @abc.abstractmethod
    def charge(self, cents: int) -> str:  # => no body -- required by every vendor adapter
        ...  # => the ellipsis stub -- concrete vendor adapters below fill this in
 
 
class StripeGateway(PaymentGateway):  # => wraps VENDOR A's own, unstable method shape
    def charge(self, cents: int) -> str:  # => defines the charge() method
        return self._stripe_create_charge(amount_cents=cents)  # => delegates to the vendor's OWN, unstable-shaped method
 
    def _stripe_create_charge(self, amount_cents: int) -> str:  # => Stripe's OWN naming/shape
        return f"stripe:charged {amount_cents} cents"  # => returns this value to the caller
 
 
class PaypalGateway(PaymentGateway):  # => wraps VENDOR B's DIFFERENTLY-shaped method entirely
    def charge(self, cents: int) -> str:  # => defines the charge() method
        dollars: float = cents / 100  # => Paypal's OWN API happens to want dollars, not cents
        return self._paypal_send_payment(dollars)  # => delegates to the vendor's OWN, differently-shaped method
 
    def _paypal_send_payment(self, dollars: float) -> str:  # => Paypal's OWN naming/shape
        return f"paypal:sent ${dollars:.2f}"  # => returns this value to the caller
 
 
class Checkout:  # => the CLIENT -- depends ONLY on PaymentGateway, never on a vendor by name
    def __init__(self, gateway: PaymentGateway) -> None:  # => the constructor
        self.gateway = gateway  # => held as the stable abstraction, swappable at construction
 
    def pay(self, cents: int) -> str:  # => defines the pay() method
        return self.gateway.charge(cents)  # => Checkout's OWN code never changes, ever
 
 
stripe_checkout: Checkout = Checkout(StripeGateway())  # => wired to vendor A
print(stripe_checkout.pay(500))  # => Checkout.pay() itself never mentions "stripe" anywhere
# => Output: stripe:charged 500 cents
 
paypal_checkout: Checkout = Checkout(PaypalGateway())  # => SWAPPED to vendor B, no Checkout edit
print(paypal_checkout.pay(500))  # => the SAME Checkout.pay() call-site, a different vendor underneath
# => Output: paypal:sent $5.00
# => Swapping StripeGateway for PaypalGateway required editing zero lines inside the Checkout class

stripe:charged 500 cents
paypal:sent $5.00

"""Example 49: pytest verification for Protected Variations Behind an Interface."""
 
from example import Checkout, PaypalGateway, StripeGateway
 
 
def test_stripe_backed_checkout_charges_through_the_vendor_adapter() -> None:
    checkout: Checkout = Checkout(StripeGateway())
    assert checkout.pay(500) == "stripe:charged 500 cents"
 
 
def test_swapping_the_vendor_requires_no_checkout_class_changes() -> None:
    # => same Checkout class, same pay() call-site, only the injected gateway differs
    checkout: Checkout = Checkout(PaypalGateway())
    assert checkout.pay(500) == "paypal:sent $5.00"
 
 
# => Run: pytest -- Output: 2 passed