Intermediate

"""Example 29: Four Ways -- Imperative."""
 
 
def word_frequency_imperative(text: str) -> dict[str, int]:  # => way #1 of 4: loop + dict, mutated in place
    counts: dict[str, int] = {}  # => mutable accumulator
    for word in text.split():  # => explicit iteration
        counts[word] = counts.get(word, 0) + 1  # => explicit mutation, one word at a time
    return counts  # => the final mutated state
 
 
sample = "red blue red green blue red"  # => shared sample text for all four "four-ways" examples
result = word_frequency_imperative(sample)  # => run it
print(result)  # => red: 3, blue: 2, green: 1
# => Output: {'red': 3, 'blue': 2, 'green': 1}

python3 example.py

{'red': 3, 'blue': 2, 'green': 1}

"""Example 29: pytest verification for Four Ways -- Imperative."""
 
from example import word_frequency_imperative
 
 
def test_counts_match_the_known_sample() -> None:
    assert word_frequency_imperative("red blue red green blue red") == {
        "red": 3,
        "blue": 2,
        "green": 1,
    }  # => shared expected result across examples 29-32
 
 
# => Run: pytest -- Output: 1 passed

pytest -q

1 passed

"""Example 30: Four Ways -- OO."""
 
 
class WordFrequencyCounter:  # => way #2 of 4: state and behavior bundled in a class
    def __init__(self) -> None:  # => constructor runs once, before count() is ever called
        self._counts: dict[str, int] = {}  # => private state, only this class's methods touch it
        # => starts empty -- every instance gets its own independent dict, never shared
 
    def count(self, text: str) -> "WordFrequencyCounter":  # => behavior: process text, mutate self
        for word in text.split():  # => same tokenization as example 29
            self._counts[word] = self._counts.get(word, 0) + 1  # => mutate this instance's own state
        return self  # => returning self allows chaining, a common OO idiom
 
    def result(self) -> dict[str, int]:  # => behavior: read this instance's own state
        return dict(self._counts)  # => defensive copy
 
 
sample = "red blue red green blue red"  # => identical sample to example 29
counter = WordFrequencyCounter().count(sample)  # => construct, then chain the count() call
print(counter.result())  # => must match example 29's dict exactly
# => Output: {'red': 3, 'blue': 2, 'green': 1}

python3 example.py

{'red': 3, 'blue': 2, 'green': 1}

"""Example 30: pytest verification for Four Ways -- OO."""
 
from example import WordFrequencyCounter
 
 
def test_oo_counts_match_the_imperative_version() -> None:
    result = WordFrequencyCounter().count("red blue red green blue red").result()
    assert result == {"red": 3, "blue": 2, "green": 1}  # => identical to example 29's result
 
 
# => Run: pytest -- Output: 1 passed

pytest -q

1 passed

"""Example 31: Four Ways -- Functional."""
 
from collections import Counter  # => way #3 of 4: a value-producing call, no visible mutation
 
sample = "red blue red green blue red"  # => identical sample to examples 29-30
result = dict(Counter(sample.split()))  # => one expression: tokenize, then fold into counts
print(result)  # => must match examples 29-30's dict exactly
# => Output: {'red': 3, 'blue': 2, 'green': 1}

python3 example.py

{'red': 3, 'blue': 2, 'green': 1}

"""Example 31: pytest verification for Four Ways -- Functional."""
 
from collections import Counter
 
 
def word_frequency_functional(text: str) -> dict[str, int]:  # => reusable helper mirroring example.py
    return dict(Counter(text.split()))  # => value-producing, no mutation of the caller's input
 
 
def test_functional_counts_match_the_other_three_ways() -> None:
    assert word_frequency_functional("red blue red green blue red") == {
        "red": 3,
        "blue": 2,
        "green": 1,
    }  # => identical to examples 29-30's result
 
 
# => Run: pytest -- Output: 1 passed

pytest -q

1 passed

"""Example 32: Four Ways -- Declarative."""
 
import sqlite3  # => the standard library's built-in SQL engine -- no external dependency needed
 
sample = "red blue red green blue red"  # => identical sample to examples 29-31
 
conn = sqlite3.connect(":memory:")  # => way #4 of 4: state the desired result, let SQLite compute it
conn.execute("CREATE TABLE words (word TEXT)")  # => declare the shape of the data
conn.executemany("INSERT INTO words VALUES (?)", [(w,) for w in sample.split()])  # => load every word
rows = conn.execute(  # => the query IS the algorithm -- no accumulator variable anywhere in this file
    "SELECT word, COUNT(*) FROM words GROUP BY word ORDER BY COUNT(*) DESC"
    # => GROUP BY + COUNT(*) declares "the frequency of each word" -- no loop mechanics anywhere
).fetchall()  # => the query planner decided HOW to group and count; this call only asks for the rows
result = dict(rows)  # => turn the declared result into the same dict shape as examples 29-31
conn.close()  # => release the connection
# => an in-memory connection's data disappears once closed -- nothing to clean up on disk
 
print(result)  # => must match examples 29-31's dict exactly, across all four paradigms
# => Output: {'red': 3, 'blue': 2, 'green': 1}

python3 example.py

{'red': 3, 'blue': 2, 'green': 1}

"""Example 32: pytest verification for Four Ways -- Declarative."""
 
import sqlite3
 
 
def word_frequency_declarative(text: str) -> dict[str, int]:  # => reusable helper mirroring example.py
    conn = sqlite3.connect(":memory:")  # => fresh in-memory database per call
    conn.execute("CREATE TABLE words (word TEXT)")
    conn.executemany("INSERT INTO words VALUES (?)", [(w,) for w in text.split()])
    rows = conn.execute("SELECT word, COUNT(*) FROM words GROUP BY word ORDER BY COUNT(*) DESC").fetchall()
    conn.close()  # => always release the connection
    return dict(rows)  # => same shape as the other three ways
 
 
def test_declarative_counts_match_all_three_other_ways() -> None:
    result = word_frequency_declarative("red blue red green blue red")
    assert result == {"red": 3, "blue": 2, "green": 1}  # => the same dict, all four paradigms agree
 
 
# => Run: pytest -- Output: 1 passed

pytest -q

1 passed

"""Example 33: State Machine -- Imperative."""
 
state = "locked"  # => MUTABLE GLOBAL: the turnstile's current state lives in one module-level box
events: list[str] = ["coin", "push", "push", "coin", "coin", "push"]  # => a sequence to replay
 
 
def handle(event: str) -> None:  # => mutates the global `state` directly, one transition-if at a time
    global state  # => explicit acknowledgement this function reaches outside itself
    if state == "locked" and event == "coin":  # => explicit transition-if #1
        state = "unlocked"  # => mutate in place
    elif state == "locked" and event == "push":  # => explicit transition-if #2
        pass  # => pushing a locked turnstile does nothing -- state stays "locked"
    elif state == "unlocked" and event == "push":  # => explicit transition-if #3
        state = "locked"  # => mutate in place
    elif state == "unlocked" and event == "coin":  # => explicit transition-if #4
        pass  # => an extra coin on an already-unlocked turnstile changes nothing
 
 
history: list[str] = [state]  # => record the state after every event, starting with the initial one
for event in events:  # => replay every event against the mutable global
    handle(event)  # => each call may mutate `state`
    history.append(state)  # => record what it became
 
print(history)  # => locked -> unlocked (coin) -> locked (push) -> ... -> locked -> unlocked
# => Output: ['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

python3 example.py

['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

"""Example 33: pytest verification for State Machine -- Imperative."""
 
import example
 
 
def test_locked_to_unlocked_to_locked_sequence() -> None:
    # => the module-level demo already replayed coin,push,push,coin,coin,push -- verify its trace
    assert example.history == ["locked", "unlocked", "locked", "locked", "unlocked", "unlocked", "locked"]
 
 
def test_pushing_a_locked_turnstile_does_not_unlock_it() -> None:
    example.state = "locked"  # => reset the shared global explicitly for this test's own run
    example.handle("push")  # => push while locked
    assert example.state == "locked"  # => must remain locked -- pushing alone never unlocks
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 34: State Machine -- OO (State Pattern)."""
 
from abc import ABC, abstractmethod  # => ABC/abstractmethod force every concrete state to implement both
 
 
class TurnstileState(ABC):  # => the State pattern: each state is its OWN object, not a string tag
    @abstractmethod  # => marks on_coin() as required -- TurnstileState itself can never be instantiated
    def on_coin(self) -> "TurnstileState":  # => every state must say what a coin does to it
        ...  # => no body here -- only concrete subclasses below provide the real behavior
 
    @abstractmethod  # => marks on_push() as required, same contract as on_coin() above
    def on_push(self) -> "TurnstileState":  # => every state must say what a push does to it
        ...  # => no body here -- only concrete subclasses below provide the real behavior
 
    name: str  # => a human-readable label, set by each concrete subclass
 
 
class Locked(TurnstileState):  # => concrete state object #1
    name = "locked"  # => satisfies the abstract `name` field declared on TurnstileState
 
    def on_coin(self) -> TurnstileState:  # => Locked's OWN answer to "what does a coin do?"
        return Unlocked()  # => transition: return a DIFFERENT state object
 
    def on_push(self) -> TurnstileState:  # => Locked's OWN answer to "what does a push do?"
        return self  # => stay locked -- return the same state object, unchanged
 
 
class Unlocked(TurnstileState):  # => concrete state object #2
    name = "unlocked"  # => satisfies the same abstract `name` field, with this state's own value
 
    def on_coin(self) -> TurnstileState:  # => Unlocked's OWN answer to a coin
        return self  # => an extra coin changes nothing
 
    def on_push(self) -> TurnstileState:  # => Unlocked's OWN answer to a push
        return Locked()  # => transition back
 
 
events: list[str] = ["coin", "push", "push", "coin", "coin", "push"]  # => same sequence as example 33
current: TurnstileState = Locked()  # => start locked, as an OBJECT, not a string
history: list[str] = [current.name]  # => record the starting state's name
 
for event in events:  # => replay the same events
    current = current.on_coin() if event == "coin" else current.on_push()  # => dispatch via the object itself
    history.append(current.name)  # => record the new state's name
 
print(history)  # => must be identical to example 33's trace
# => no `if state == "locked"` anywhere -- transitions live inside each state object's own methods
# => Output: ['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

python3 example.py

['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

"""Example 34: pytest verification for State Machine -- OO (State Pattern)."""
 
from example import Locked, TurnstileState, Unlocked
 
 
def test_state_pattern_trace_matches_the_imperative_version() -> None:
    events = ["coin", "push", "push", "coin", "coin", "push"]  # => same sequence as example 33
    current: TurnstileState = Locked()  # => start locked
    history = [current.name]
    for event in events:
        current = current.on_coin() if event == "coin" else current.on_push()
        history.append(current.name)
    assert history == ["locked", "unlocked", "locked", "locked", "unlocked", "unlocked", "locked"]
 
 
def test_each_transition_returns_a_distinct_state_object() -> None:
    locked = Locked()  # => construct once
    unlocked = locked.on_coin()  # => transition via a coin
    assert isinstance(unlocked, Unlocked)  # => coin from Locked always yields an Unlocked object
    assert unlocked.on_push().name == "locked"  # => push from Unlocked always yields back to locked
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 35: State Machine -- Functional."""
 
from functools import reduce  # => reduce() is Python's built-in fold: combine a sequence into one value
 
 
def transition(state: str, event: str) -> str:  # => a PURE function: (state, event) -> new state
    if state == "locked" and event == "coin":  # => same rules as examples 33-34, expressed as pure data-in data-out
        return "unlocked"  # => no assignment to any outer variable -- just a returned value
    if state == "unlocked" and event == "push":  # => the other real transition
        return "locked"  # => same shape: a returned value, nothing mutated
    return state  # => every other combination is a no-op -- return the SAME state, no mutation anywhere
 
 
events: list[str] = ["coin", "push", "push", "coin", "coin", "push"]  # => same sequence as examples 33-34
 
# => a FOLD builds the whole history in one expression -- no loop body visibly mutates anything
history = reduce(  # => reduce(fn, sequence, initial) threads an accumulator through every event
    lambda states, event: states + [transition(states[-1], event)],  # => append the next state, functionally
    events,  # => the sequence being folded over, one event per step
    ["locked"],  # => the fold's starting accumulator: history begins with just the initial state
)  # => the final accumulator value IS the complete, fully-built history
 
print(history)  # => must be identical to examples 33-34's trace
# => Output: ['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

python3 example.py

['locked', 'unlocked', 'locked', 'locked', 'unlocked', 'unlocked', 'locked']

"""Example 35: pytest verification for State Machine -- Functional."""
 
from example import transition
 
 
def test_pure_transition_matches_the_other_two_versions() -> None:
    from functools import reduce
 
    events = ["coin", "push", "push", "coin", "coin", "push"]  # => same sequence as examples 33-34
    history = reduce(lambda states, event: states + [transition(states[-1], event)], events, ["locked"])
    assert history == ["locked", "unlocked", "locked", "locked", "unlocked", "unlocked", "locked"]
 
 
def test_transition_never_mutates_its_string_arguments() -> None:
    before_state, before_event = "locked", "coin"  # => strings are immutable in Python regardless,
    result = transition(before_state, before_event)  # => but this documents the pure-function contract
    assert before_state == "locked" and before_event == "coin"  # => arguments are provably unchanged
    assert result == "unlocked"  # => and the correct new state was returned as a NEW value
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 36: Prolog-in-Python (Unification + Backtracking)."""
 
from collections.abc import Iterator  # => grandparent() below is a generator, typed as an Iterator
 
Fact = tuple[str, str]  # => a (parent, child) fact, mirroring example 19's family data
facts: list[Fact] = [  # => the FACTS -- raw data, no rule about grandparents anywhere in this list
    ("alice", "bob"),  # => alice is bob's parent
    ("bob", "carol"),  # => bob is carol's parent
    ("carol", "dave"),  # => carol is dave's parent
]  # => the same three-generation family as example 19, this time queried via search, not a comprehension
 
 
def parent(x: str, y: str) -> bool:  # => the base relation: is (x, y) a stored fact? (unification step)
    return (x, y) in facts  # => "unifying" x and y against every stored fact
 
 
def grandparent(x: str) -> Iterator[str]:  # => the RULE, expressed as a search over intermediate variables
    for _px, y in facts:  # => try binding Y to every fact's child (a backtracking choice point)
        if _px != x:  # => this choice point only matters when x is actually the parent in this fact
            continue  # => BACKTRACK: this binding of Y didn't unify with parent(x, Y) -- try the next one
        for py, z in facts:  # => a NESTED choice point: try binding Z to every fact's child
            if py != y:  # => does this fact's parent match the Y we just bound?
                continue  # => BACKTRACK again: try the next candidate fact
            yield z  # => both parent(x, Y) and parent(Y, Z) unified -- z is a valid answer
 
 
results = list(grandparent("alice"))  # => search: alice -> bob (bind Y) -> bob -> carol (bind Z)
# => draining the generator forces the backtracking search to actually run to completion
print(results)  # => must match example 19's comprehension-based answer
# => Output: ['carol']

python3 example.py

['carol']

"""Example 36: pytest verification for Prolog-in-Python (Unification + Backtracking)."""
 
from example import grandparent
 
 
def test_grandparent_query_resolves_via_search() -> None:
    assert list(grandparent("alice")) == ["carol"]  # => same answer as example 19's comprehension version
 
 
def test_a_person_with_no_grandchildren_yields_nothing() -> None:
    assert list(grandparent("dave")) == []  # => dave has no recorded children at all -- search finds none
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 37: Backtracking N-Queens."""
 
 
def solve_n_queens(n: int) -> list[int] | None:  # => returns one solution: cols[row] = column of the queen
    cols: list[int] = []  # => the partial (and eventually full) placement, one column index per row
 
    def is_safe(row: int, col: int) -> bool:  # => can a queen go at (row, col) given queens placed so far?
        for placed_row, placed_col in enumerate(cols):  # => check against every already-placed queen
            if placed_col == col:  # => same column: an attack
                return False  # => reject immediately -- no need to check any other placed queen
            if abs(placed_col - col) == abs(placed_row - row):  # => same diagonal: an attack
                return False  # => reject immediately, same reasoning
        return True  # => no earlier queen attacks this square
 
    def backtrack(row: int) -> bool:  # => try to place a queen in every row, from `row` downward
        if row == n:  # => base case: every row has a queen -- a full solution was found
            return True  # => success propagates straight back up the recursion, no cols.pop() needed
        for col in range(n):  # => CHOICE POINT: try every column in this row
            if is_safe(row, col):  # => only attempt columns that don't conflict yet
                cols.append(col)  # => tentatively place the queen
                if backtrack(row + 1):  # => recurse to the next row
                    return True  # => the whole rest of the board solved -- propagate success up
                cols.pop()  # => BACKTRACK: that column led nowhere, undo it and try the next column
        return False  # => no column in this row works given the current partial placement
 
    return cols if backtrack(0) else None  # => cols is fully built only if backtrack(0) succeeded
 
 
def no_two_queens_attack(cols: list[int]) -> bool:  # => independent checker, used only for verification
    for r1 in range(len(cols)):  # => compare every pair of placed queens
        for r2 in range(r1 + 1, len(cols)):  # => r2 > r1 -- each pair checked exactly once, not twice
            if cols[r1] == cols[r2]:  # => same column
                return False  # => a violation was found -- no need to check any remaining pairs
            if abs(cols[r1] - cols[r2]) == abs(r1 - r2):  # => same diagonal
                return False  # => a violation was found -- no need to check any remaining pairs
    return True  # => every pair is safe
 
 
solution = solve_n_queens(8)  # => the classic 8-queens problem
assert solution is not None  # => narrow away None -- 8-queens always has a solution, matching test_example.py
print(solution)  # => one valid arrangement (the specific columns depend on search order, but it is safe)
# => Output: [0, 4, 7, 5, 2, 6, 1, 3]
print(no_two_queens_attack(solution))  # => independently confirms no two queens attack each other
# => Output: True

python3 example.py

[0, 4, 7, 5, 2, 6, 1, 3]
True

"""Example 37: pytest verification for Backtracking N-Queens."""
 
from example import no_two_queens_attack, solve_n_queens
 
 
def test_eight_queens_solution_has_no_attacking_pair() -> None:
    solution = solve_n_queens(8)  # => same size as the module-level demo
    assert solution is not None  # => a solution must exist for n=8
    assert len(solution) == 8  # => one queen per row
    assert no_two_queens_attack(solution)  # => the independent checker must confirm safety
 
 
def test_four_queens_also_solves_safely() -> None:
    solution = solve_n_queens(4)  # => a smaller board, still solvable
    assert solution is not None
    assert no_two_queens_attack(solution)
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 38: Constraint Map Coloring."""
 
Region = str  # => a type alias -- purely for readability, region names are just strings
Color = str  # => a type alias -- purely for readability, color names are just strings
 
adjacency: dict[Region, list[Region]] = {  # => DECLARE the constraint: which regions must differ in color
    "west": ["central"],  # => west is adjacent to central only
    "central": ["west", "east"],  # => central is adjacent to both neighbors
    "east": ["central"],  # => east is adjacent to central only
}  # => nothing here says HOW to search -- just which pairs may never share a color
 
colors: list[Color] = ["red", "green", "blue"]  # => the available palette (3-coloring)
 
 
def solve_coloring(adj: dict[Region, list[Region]], palette: list[Color]) -> dict[Region, Color] | None:  # => the GENERIC part: works for any adjacency map, any palette
    regions = list(adj.keys())  # => a fixed order to assign regions in
    assignment: dict[Region, Color] = {}  # => the partial (then full) solution being built
 
    def backtrack(index: int) -> bool:  # => try to color every region from `index` onward
        if index == len(regions):  # => base case: every region has a color -- solved
            return True  # => the assignment dict already holds a complete, valid coloring
        region = regions[index]  # => the region we're choosing a color for right now
        for color in palette:  # => CHOICE POINT: try every color in the palette
            if all(assignment.get(neighbor) != color for neighbor in adj[region]):  # => check the constraint
                assignment[region] = color  # => tentatively assign
                if backtrack(index + 1):  # => recurse to the next region
                    return True  # => success propagates straight back up -- no del assignment[region] needed
                del assignment[region]  # => BACKTRACK: undo this color, try the next one
        return False  # => no color in the palette works given the current partial assignment
 
    return dict(assignment) if backtrack(0) else None  # => copy out only on success
 
 
result = solve_coloring(adjacency, colors)  # => run the solver
assert result is not None  # => narrow away None -- this adjacency/palette pair always has a valid coloring
print(result)  # => west and east may share a color; central must differ from both
# => Output: {'west': 'red', 'central': 'green', 'east': 'red'}
print(all(result[a] != result[b] for a, neighbors in adjacency.items() for b in neighbors))  # => verify
# => Output: True

python3 example.py

{'west': 'red', 'central': 'green', 'east': 'red'}
True

"""Example 38: pytest verification for Constraint Map Coloring."""
 
from example import solve_coloring
 
 
def test_no_adjacent_regions_share_a_color() -> None:
    adjacency = {"west": ["central"], "central": ["west", "east"], "east": ["central"]}
    colors = ["red", "green", "blue"]
    result = solve_coloring(adjacency, colors)
    assert result is not None  # => a valid 3-coloring must exist for this simple adjacency graph
    for region, neighbors in adjacency.items():  # => check every declared constraint holds
        for neighbor in neighbors:
            assert result[region] != result[neighbor]  # => the core map-coloring constraint
 
 
def test_two_colors_are_insufficient_for_a_triangle() -> None:
    triangle = {"a": ["b", "c"], "b": ["a", "c"], "c": ["a", "b"]}  # => every region touches both others
    assert solve_coloring(triangle, ["red", "green"]) is None  # => a triangle needs at least 3 colors
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 39: Constraint Mini Sudoku (4x4)."""
 
Board = list[list[int]]  # => a 4x4 grid; 0 marks an empty cell
 
# => DECLARE the puzzle: 0 = empty, otherwise a fixed clue -- one valid 4x4 sudoku with three clues
puzzle: Board = [  # => four rows, four clues total -- the rest is left for the solver to fill in
    [1, 0, 0, 0],  # => row 0: a 1 fixed in column 0
    [0, 0, 1, 0],  # => row 1: a 1 fixed in column 2
    [0, 1, 0, 0],  # => row 2: a 1 fixed in column 1
    [0, 0, 0, 1],  # => row 3: a 1 fixed in column 3
]  # => closes the puzzle's initial 4x4 grid
 
 
def box_id(row: int, col: int) -> tuple[int, int]:  # => which 2x2 box a cell belongs to
    return (row // 2, col // 2)  # => integer division groups rows/cols into 2x2 quadrants
 
 
def is_valid(board: Board, row: int, col: int, value: int) -> bool:  # => the three sudoku constraints
    if any(board[row][c] == value for c in range(4)):  # => row constraint: no repeat in the row
        return False  # => reject immediately -- no need to check column or box constraints
    if any(board[r][col] == value for r in range(4)):  # => column constraint: no repeat in the column
        return False  # => reject immediately -- no need to check the box constraint
    br, bc = box_id(row, col)  # => box constraint: no repeat in the same 2x2 box
    for r in range(br * 2, br * 2 + 2):  # => the two rows of this cell's own 2x2 box
        for c in range(bc * 2, bc * 2 + 2):  # => the two columns of this cell's own 2x2 box
            if board[r][c] == value:  # => a same-box cell already holds this value
                return False  # => reject -- the value would appear twice in the same box
    return True  # => all three constraints satisfied
 
 
def solve(board: Board) -> Board | None:  # => backtracking search over empty cells
    for row in range(4):  # => find the first empty cell, in reading order
        for col in range(4):  # => scan every column of this row before moving to the next row
            if board[row][col] == 0:  # => this is the next cell to fill
                for value in range(1, 5):  # => CHOICE POINT: try every candidate digit 1-4
                    if is_valid(board, row, col, value):  # => only try digits that satisfy all constraints
                        board[row][col] = value  # => tentatively place it
                        if solve(board):  # => recurse into the rest of the board
                            return board  # => success propagates straight up -- no undo needed here
                        board[row][col] = 0  # => BACKTRACK: undo, try the next candidate digit
                return None  # => no digit worked for this cell given the current partial board
    return board  # => no empty cells remain -- fully solved
 
 
solution = solve([row[:] for row in puzzle])  # => solve a COPY so the original `puzzle` stays untouched
assert solution is not None  # => narrow away None -- this puzzle's three clues always admit a solution
print(solution)  # => a fully filled, constraint-satisfying 4x4 grid
# => Output: [[1, 2, 3, 4], [3, 4, 1, 2], [2, 1, 4, 3], [4, 3, 2, 1]]
rows_ok = all(sorted(row) == [1, 2, 3, 4] for row in solution)  # => every row has 1-4 exactly once
cols_ok = all(sorted(solution[r][c] for r in range(4)) == [1, 2, 3, 4] for c in range(4))  # => every column
print(rows_ok and cols_ok)  # => independently confirms rows and columns are valid
# => Output: True

python3 example.py

[[1, 2, 3, 4], [3, 4, 1, 2], [2, 1, 4, 3], [4, 3, 2, 1]]
True

"""Example 39: pytest verification for Constraint Mini Sudoku (4x4)."""
 
from example import box_id, solve
 
 
def test_solved_board_has_valid_rows_columns_and_boxes() -> None:
    puzzle = [[1, 0, 0, 0], [0, 0, 1, 0], [0, 1, 0, 0], [0, 0, 0, 1]]  # => same puzzle as the demo
    solution = solve([row[:] for row in puzzle])  # => solve a defensive copy
    assert solution is not None  # => this puzzle is solvable
    for row in solution:  # => row constraint
        assert sorted(row) == [1, 2, 3, 4]
    for col in range(4):  # => column constraint
        assert sorted(solution[r][col] for r in range(4)) == [1, 2, 3, 4]
    boxes: dict[tuple[int, int], list[int]] = {}  # => box constraint
    for r in range(4):
        for c in range(4):
            boxes.setdefault(box_id(r, c), []).append(solution[r][c])
    for values in boxes.values():
        assert sorted(values) == [1, 2, 3, 4]
 
 
def test_original_puzzle_is_not_mutated_by_solving_a_copy() -> None:
    puzzle = [[1, 0, 0, 0], [0, 0, 1, 0], [0, 1, 0, 0], [0, 0, 0, 1]]  # => fresh puzzle for this test
    before = [row[:] for row in puzzle]  # => snapshot before solving
    solve([row[:] for row in puzzle])  # => solve a copy, discard the result
    assert puzzle == before  # => the original puzzle list is byte-identical to its snapshot
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 40: Event-Driven Loop."""
 
from collections import deque  # => deque gives O(1) popleft(), the FIFO queue's core operation
from collections.abc import Callable  # => types every handler stored in the routing table below
from dataclasses import dataclass  # => @dataclass auto-generates Event's __init__ from its fields
 
 
@dataclass  # => auto-generates Event's __init__ from the two fields below
class Event:  # => a plain data record describing what happened
    kind: str  # => which handler should process this event, e.g. "login"
    payload: str  # => the data the handler needs, e.g. a username
 
 
processed: list[str] = []  # => records the order events were actually handled in
 
 
def on_login(event: Event) -> None:  # => handler for "login" events
    processed.append(f"login:{event.payload}")  # => the framework calls this -- it never calls the loop
 
 
def on_logout(event: Event) -> None:  # => handler for "logout" events
    processed.append(f"logout:{event.payload}")  # => same shape as on_login, different kind and record
 
 
handlers: dict[str, Callable[[Event], None]] = {"login": on_login, "logout": on_logout}  # => routing table
 
queue: deque[Event] = deque(  # => a FIFO queue -- events wait here until the loop drains them
    [  # => the events, already enqueued in the order they should be processed
        Event("login", "alice"),  # => first to arrive, first to be handled
        Event("login", "bob"),  # => second to arrive
        Event("logout", "alice"),  # => third to arrive
    ]  # => closes the initial list of queued events
)  # => closes the deque(...) constructor call
 
while queue:  # => the event loop: keep draining until the queue is empty
    event = queue.popleft()  # => take the OLDEST event first -- FIFO order
    handlers[event.kind](event)  # => route it to its handler and run that handler NOW
    # => the loop itself decides WHEN each handler runs -- the handler never calls back into the loop
 
print(processed)  # => events must be processed in the exact order they were enqueued
# => Output: ['login:alice', 'login:bob', 'logout:alice']

python3 example.py

['login:alice', 'login:bob', 'logout:alice']

"""Example 40: pytest verification for Event-Driven Loop."""
 
from collections import deque
from collections.abc import Callable
 
from example import Event
 
 
def test_events_processed_in_fifo_order() -> None:
    seen: list[str] = []  # => local recorder, isolated from the module-level demo
    handlers: dict[str, Callable[[Event], None]] = {
        "login": lambda e: seen.append(f"login:{e.payload}"),
        "logout": lambda e: seen.append(f"logout:{e.payload}"),
    }
    queue = deque([Event("login", "x"), Event("logout", "y"), Event("login", "z")])
    while queue:
        event = queue.popleft()
        handlers[event.kind](event)
    assert seen == ["login:x", "logout:y", "login:z"]  # => exactly the enqueue order, unchanged
 
 
def test_named_handlers_produce_the_documented_module_level_trace() -> None:
    from example import processed  # => the module-level demo already ran when example.py was imported
 
    assert processed == ["login:alice", "login:bob", "logout:alice"]  # => matches example.py's own Output
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 41: Reactive Derived Value."""
 
from collections.abc import Callable  # => types every no-argument callback stored below
 
 
class Signal:  # => a reactive source value that notifies dependents automatically
    def __init__(self, initial: int) -> None:  # => constructor seeds the starting value
        self._value = initial  # => the current value, hidden behind get()/set() below
        self._on_change: list[Callable[[], None]] = []  # => callbacks to run whenever this signal changes
 
    def get(self) -> int:  # => read the current value
        return self._value  # => a plain read -- getting never triggers propagation
 
    def set(self, value: int) -> None:  # => write a new value and PUSH the change to every dependent
        self._value = value  # => update the internal box first
        for callback in self._on_change:  # => automatically notify -- no dependent has to poll
            callback()  # => runs the dependent's recompute hook synchronously, right here
 
    def on_change(self, callback: Callable[[], None]) -> None:  # => register a dependent's recompute hook
        self._on_change.append(callback)  # => append only -- does NOT call callback with the current value
 
 
class Computed:  # => a derived signal: recomputes automatically whenever a source changes
    def __init__(self, compute: Callable[[], int], *sources: Signal) -> None:  # => wires up every source
        self._compute = compute  # => the formula, e.g. "a.get() + b.get()"
        self.value = compute()  # => compute once immediately, so `c` is correct before any update
        for source in sources:  # => subscribe to EVERY source this computed value depends on
            source.on_change(self._recompute)  # => wire automatic propagation
 
    def _recompute(self) -> None:  # => runs automatically whenever ANY source signal changes
        self.value = self._compute()  # => re-run the formula and refresh the cached value
 
 
a = Signal(1)  # => source signal a
b = Signal(2)  # => source signal b
c = Computed(lambda: a.get() + b.get(), a, b)  # => c = a + b, kept up to date automatically
 
print(c.value)  # => 1 + 2, computed at construction time
# => Output: 3
a.set(10)  # => changing a automatically triggers c's recompute -- no manual "update c" call needed
print(c.value)  # => 10 + 2
# => Output: 12
b.set(20)  # => changing b ALSO automatically triggers c's recompute
print(c.value)  # => 10 + 20
# => Output: 30

python3 example.py

3
12
30

"""Example 41: pytest verification for Reactive Derived Value."""
 
from example import Computed, Signal
 
 
def test_computed_value_after_two_updates() -> None:
    a = Signal(1)  # => fresh signals, isolated from the module-level demo
    b = Signal(2)
    c = Computed(lambda: a.get() + b.get(), a, b)
    assert c.value == 3  # => 1 + 2 at construction
 
    a.set(10)  # => update source a
    assert c.value == 12  # => c recomputed automatically: 10 + 2
 
    b.set(20)  # => update source b too
    assert c.value == 30  # => c recomputed again: 10 + 20
 
 
def test_computed_never_needs_a_manual_recompute_call() -> None:
    a = Signal(0)  # => a fresh independent pair of signals
    b = Signal(0)
    c = Computed(lambda: a.get() * b.get(), a, b)
    a.set(5)
    b.set(4)
    assert c.value == 20  # => 5 * 4, with no explicit `c.recompute()` call anywhere in this test
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 42: Reactive vs Manual Recompute."""
 
from collections.abc import Callable  # => types every no-argument callback stored below
 
 
class ManualPair:  # => BEFORE: caller must REMEMBER to update the derived value by hand
    def __init__(self, a: int, b: int) -> None:  # => constructor seeds both inputs and the derived total
        self.a = a  # => plain field, no notification wiring at all
        self.b = b  # => plain field, no notification wiring at all
        self.total = a + b  # => computed once -- nothing keeps this in sync automatically
 
    def set_a(self, value: int) -> None:  # => updates a but does NOT touch total -- easy to forget
        self.a = value  # => total is now STALE until someone remembers to call recompute_total()
 
    def recompute_total(self) -> None:  # => the easy-to-forget manual step
        self.total = self.a + self.b  # => must be called explicitly -- nothing calls it automatically
 
 
class Signal:  # => AFTER: the same minimal reactive primitive as example 41
    def __init__(self, initial: int) -> None:  # => constructor seeds the starting value
        self._value = initial  # => the current value, hidden behind get()/set() below
        self._on_change: list[Callable[[], None]] = []  # => callbacks to run whenever this signal changes
 
    def get(self) -> int:  # => read the current value
        return self._value  # => a plain read -- getting never triggers propagation
 
    def set(self, value: int) -> None:  # => setting AUTOMATICALLY notifies every dependent
        self._value = value  # => update the internal box first
        for callback in self._on_change:  # => automatically notify -- no dependent has to poll
            callback()  # => runs the dependent's recompute hook synchronously, right here
 
    def on_change(self, callback: Callable[[], None]) -> None:  # => register a dependent's recompute hook
        self._on_change.append(callback)  # => append only -- does NOT call callback with the current value
 
 
class ReactivePair:  # => wires a and b so total NEVER goes stale
    def __init__(self, a: int, b: int) -> None:  # => constructor wraps both inputs as Signals
        self.a = Signal(a)  # => a is now reactive, not a plain field
        self.b = Signal(b)  # => b is now reactive, not a plain field
        self.total = self.a.get() + self.b.get()  # => initial value
        self.a.on_change(self._recompute)  # => subscribe -- total tracks a automatically
        self.b.on_change(self._recompute)  # => subscribe -- total tracks b automatically
 
    def _recompute(self) -> None:  # => runs automatically whenever a or b changes
        self.total = self.a.get() + self.b.get()  # => the SAME formula as ManualPair, but never forgotten
 
 
manual = ManualPair(1, 2)  # => BEFORE
manual.set_a(10)  # => forgot to call recompute_total() -- a realistic mistake
print(manual.total)  # => STALE: still reflects the OLD a, not the new one
# => Output: 3
 
reactive = ReactivePair(1, 2)  # => AFTER
reactive.a.set(10)  # => the equivalent update, via the reactive API
print(reactive.total)  # => automatically current: 10 + 2
# => Output: 12

python3 example.py

3
12

"""Example 42: pytest verification for Reactive vs Manual Recompute."""
 
from example import ManualPair, ReactivePair
 
 
def test_manual_pair_goes_stale_if_recompute_is_forgotten() -> None:
    pair = ManualPair(1, 2)  # => fresh instance
    pair.set_a(10)  # => update a, but never call recompute_total()
    assert pair.total == 3  # => STILL the old total -- this is the bug the reactive version prevents
 
 
def test_manual_pair_is_correct_only_after_an_explicit_recompute() -> None:
    pair = ManualPair(1, 2)  # => fresh instance
    pair.set_a(10)  # => update a
    pair.recompute_total()  # => remember to call it this time
    assert pair.total == 12  # => now correct, but only because of the explicit call
 
 
def test_reactive_pair_is_always_consistent_automatically() -> None:
    pair = ReactivePair(1, 2)  # => fresh instance
    pair.a.set(10)  # => the same conceptual update, no manual recompute anywhere
    assert pair.total == 12  # => correct immediately -- reactive propagation is automatic
 
 
# => Run: pytest -- Output: 3 passed

pytest -q

3 passed

"""Example 43: Dataflow Topological Execute."""
 
from collections.abc import Callable  # => types every node's compute formula stored below
 
Node = str  # => a type alias -- node names are just strings
graph: dict[Node, list[Node]] = {  # => DAG of value dependencies: node -> nodes it depends on
    "c": ["a", "b"],  # => c depends on a and b
    "b": ["a"],  # => b depends on a
    "a": [],  # => a has no dependencies -- a source node
}  # => closes the dependency graph declaration
formulas: dict[Node, Callable[[dict[Node, int]], int]] = {  # => how to compute each node, given results so far
    "a": lambda results: 1,  # => a is a constant
    "b": lambda results: results["a"] + 1,  # => b = a + 1
    "c": lambda results: results["a"] + results["b"],  # => c = a + b
}  # => closes the per-node formula table
 
 
def topological_order(deps: dict[Node, list[Node]]) -> list[Node]:  # => order respecting every dependency
    visited: set[Node] = set()  # => nodes already placed in the order
    order: list[Node] = []  # => the order being built
 
    def visit(node: Node) -> None:  # => depth-first visit: dependencies before the node itself
        if node in visited:  # => already placed -- nothing to do
            return  # => stops the recursion for this branch -- a node is never appended twice
        visited.add(node)  # => mark BEFORE recursing to guard against revisiting a node mid-traversal
        for dep in deps[node]:  # => every dependency must appear in the order first
            visit(dep)  # => recurse into the dependency
        order.append(node)  # => only append AFTER all dependencies are already in `order`
 
    for node in deps:  # => make sure every node gets visited, regardless of starting point
        visit(node)  # => already-visited nodes short-circuit immediately via the check above
    return order  # => a valid topological order: every dependency precedes its dependents
 
 
order = topological_order(graph)  # => compute the execution order
print(order)  # => a must come before b and c; b must come before c
# => Output: ['a', 'b', 'c']
 
results: dict[Node, int] = {}  # => accumulate computed values as we execute in order
for node in order:  # => execute strictly in topological order
    results[node] = formulas[node](results)  # => by the time we reach a node, its deps are already computed
print(results)  # => a=1, b=1+1=2, c=1+2=3
# => Output: {'a': 1, 'b': 2, 'c': 3}

python3 example.py

['a', 'b', 'c']
{'a': 1, 'b': 2, 'c': 3}

"""Example 43: pytest verification for Dataflow Topological Execute."""
 
from collections.abc import Callable
 
from example import topological_order
 
 
def test_order_respects_every_dependency() -> None:
    graph = {"c": ["a", "b"], "b": ["a"], "a": []}  # => same graph as the module-level demo
    order = topological_order(graph)
    assert order.index("a") < order.index("b")  # => a must precede b
    assert order.index("a") < order.index("c")  # => a must precede c
    assert order.index("b") < order.index("c")  # => b must precede c
 
 
def test_result_matches_the_documented_formulas() -> None:
    graph = {"c": ["a", "b"], "b": ["a"], "a": []}
    formulas: dict[str, Callable[[dict[str, int]], int]] = {
        "a": lambda r: 1,
        "b": lambda r: r["a"] + 1,
        "c": lambda r: r["a"] + r["b"],
    }
    order = topological_order(graph)
    results: dict[str, int] = {}
    for node in order:
        results[node] = formulas[node](results)
    assert results == {"a": 1, "b": 2, "c": 3}  # => matches example.py's own Output exactly
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 44: Generator Pull Pipeline."""
 
from collections.abc import Callable, Iterator  # => every function below is typed as a lazy, pull-based Iterator
 
computed_log: list[int] = []  # => records every value the source generator actually produced
 
 
def source() -> Iterator[int]:  # => an "infinite" source -- would hang if fully consumed eagerly
    n = 0  # => starts at 0, incremented once per pull
    while True:  # => never terminates on its own -- laziness is the only thing that makes this safe
        n += 1  # => the next candidate value
        computed_log.append(n)  # => record that this value was actually generated (proves pull, not push)
        yield n  # => PULL-based: this line only runs when something asks the generator for its next value
 
 
def gen_map(it: Iterator[int], fn: Callable[[int], int]) -> Iterator[int]:  # => lazy map: transforms values ONE AT A TIME, on demand
    for value in it:  # => pulling from `it` only happens as this generator itself is pulled from
        yield fn(value)  # => nothing is computed until a consumer asks for the next item
 
 
def gen_filter(it: Iterator[int], predicate: Callable[[int], bool]) -> Iterator[int]:  # => lazy filter: same pull-based contract
    for value in it:  # => each pull here triggers exactly one pull upstream
        if predicate(value):  # => only values passing the predicate are ever yielded downstream
            yield value  # => only yield values that pass the predicate
 
 
def take(it: Iterator[int], n: int) -> list[int]:  # => the ONLY thing that actually drives the pipeline
    result: list[int] = []  # => the concrete list being built, one pull at a time
    for value in it:  # => pulling n times cascades back through filter -> map -> source
        result.append(value)  # => record this match before checking whether we have enough yet
        if len(result) == n:  # => stop pulling the instant we have enough -- laziness in action
            break  # => no further pulls happen -- the upstream generators simply stay paused
    return result  # => exactly n items, and not one pull more than strictly needed to produce them
 
 
pipeline = gen_filter(gen_map(source(), lambda n: n * n), lambda n: n % 2 == 0)  # => squares, then evens only
result = take(pipeline, 3)  # => pull exactly 3 matching items -- nothing more
 
print(result)  # => squares of 1..: 1,4,9,16,25,36,... ; even ones in order: 4, 16, 36
# => Output: [4, 16, 36]
print(len(computed_log))  # => the source only ran as many times as strictly needed to produce 3 matches
# => Output: 6

python3 example.py

[4, 16, 36]
6

"""Example 44: pytest verification for Generator Pull Pipeline."""
 
from example import gen_filter, gen_map, source, take
 
 
def test_first_three_even_squares() -> None:
    pipeline = gen_filter(gen_map(source(), lambda n: n * n), lambda n: n % 2 == 0)
    assert take(pipeline, 3) == [4, 16, 36]  # => same result as example.py's own Output
 
 
def test_pipeline_only_pulls_as_many_source_values_as_strictly_needed() -> None:
    seen: list[int] = []  # => a fresh local counter, isolated from the module-level demo's `computed_log`
 
    def counting_source():
        n = 0
        while True:
            n += 1
            seen.append(n)
            yield n
 
    pipeline = gen_filter(gen_map(counting_source(), lambda n: n * n), lambda n: n % 2 == 0)
    take(pipeline, 1)  # => only ask for ONE matching item
    assert seen == [1, 2]  # => n=1 (square 1, odd, rejected), n=2 (square 4, even, accepted) -- then stop
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 45: Inversion of Control."""
 
from collections.abc import Callable  # => types the row-transforming function used in both styles
 
 
def render_report_you_call_library(rows: list[str], formatter: Callable[[str], str]) -> list[str]:  # => plain function -- callers drive it directly
    # => YOU-CALL-LIBRARY: your code drives the loop, YOU decide when to call the library function
    return [formatter(row) for row in rows]  # => your code is in charge of the control flow
 
 
class ReportFramework:  # => FRAMEWORK-CALLS-YOU: the framework owns the loop, you just supply a hook
    def __init__(self) -> None:  # => constructor starts with no handler registered yet
        self._on_row: Callable[[str], str] | None = None  # => a slot for YOUR callback
 
    def register(self, handler: Callable[[str], str]) -> None:  # => you hand the framework your logic
        self._on_row = handler  # => the framework stores it, does not call it yet
 
    def run(self, rows: list[str]) -> list[str]:  # => the framework owns this loop entirely
        assert self._on_row is not None, "must register a handler first"  # => fail loudly if nothing was registered
        return [self._on_row(row) for row in rows]  # => the FRAMEWORK calls YOUR code, not the other way
 
 
rows = ["alice", "bob"]  # => shared sample data
shout: Callable[[str], str] = lambda row: row.upper()  # noqa: E731  # => the same transformation logic in both styles
 
you_call_result = render_report_you_call_library(rows, shout)  # => your code drives the call
print(you_call_result)  # => both styles must produce identical output
# => Output: ['ALICE', 'BOB']
 
framework = ReportFramework()  # => construct the framework
framework.register(shout)  # => hand it your handler -- inversion of control: framework decides when to call it
framework_result = framework.run(rows)  # => the framework's run() loop is what actually invokes `shout`
print(framework_result)  # => must be identical to the you-call-library result
# => Output: ['ALICE', 'BOB']
print(you_call_result == framework_result)  # => confirms both control-flow styles agree
# => Output: True

python3 example.py

['ALICE', 'BOB']
['ALICE', 'BOB']
True

"""Example 45: pytest verification for Inversion of Control."""
 
from collections.abc import Callable
 
from example import ReportFramework, render_report_you_call_library
 
 
def test_you_call_library_and_framework_calls_you_agree() -> None:
    rows = ["x", "y", "z"]  # => fresh sample, isolated from the module-level demo
    handler: Callable[[str], str] = lambda row: row.upper()  # noqa: E731
    direct = render_report_you_call_library(rows, handler)  # => your code drives the loop
 
    framework = ReportFramework()
    framework.register(handler)  # => hand the SAME handler to the framework
    inverted = framework.run(rows)  # => the framework drives the loop this time
 
    assert direct == inverted == ["X", "Y", "Z"]  # => identical results, different control-flow owner
 
 
def test_framework_invokes_the_registered_handler_not_a_default() -> None:
    framework = ReportFramework()  # => fresh framework instance
    calls: list[str] = []  # => records what the registered handler actually received
    framework.register(lambda row: calls.append(row) or row)  # => a handler with a visible side effect
    framework.run(["p", "q"])  # => the framework is the one that calls it, per row
    assert calls == ["p", "q"]  # => confirms the framework actually invoked OUR handler, in order
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 46: Declarative Config vs Setup."""
 
from dataclasses import dataclass, field  # => @dataclass generates __init__; field() gives a fresh list
 
 
@dataclass  # => auto-generates Server's __init__ from the three fields below
class Server:  # => the object both styles below must end up constructing, identically
    host: str = "localhost"  # => default value, overridden by both build functions below
    port: int = 8080  # => default value, overridden by both build functions below
    routes: list[str] = field(default_factory=list[str])  # => default: a fresh empty list per instance
 
 
def build_via_imperative_setup() -> Server:  # => HOW: step-by-step mutation after construction
    server = Server()  # => start from the defaults
    server.host = "api.example.com"  # => step 1: mutate host
    server.port = 443  # => step 2: mutate port
    server.routes.append("/health")  # => step 3: mutate routes
    server.routes.append("/users")  # => step 4: mutate routes again
    return server  # => the fully-mutated object
 
 
DECLARED_SPEC: dict[str, object] = {  # => WHAT: the desired final shape, stated as data up front
    "host": "api.example.com",  # => same value the imperative version reaches via mutation
    "port": 443,  # => same value the imperative version reaches via mutation
    "routes": ["/health", "/users"],  # => same value the imperative version reaches via two .append() calls
}  # => closes the declared spec -- one value, no steps
 
 
def build_via_declared_spec(spec: dict[str, object]) -> Server:  # => construct directly FROM the spec
    return Server(host=str(spec["host"]), port=int(spec["port"]), routes=list(spec["routes"]))  # => reads the whole desired shape from spec in one call  # type: ignore[arg-type]
    # => one call, reading the entire desired shape from a single declared value
 
 
imperative_server = build_via_imperative_setup()  # => run the step-by-step version
declarative_server = build_via_declared_spec(DECLARED_SPEC)  # => run the declared-spec version
 
print(imperative_server)  # => both objects must be field-for-field equal
# => Output: Server(host='api.example.com', port=443, routes=['/health', '/users'])
print(imperative_server == declarative_server)  # => dataclasses compare structurally by default
# => Output: True

python3 example.py

Server(host='api.example.com', port=443, routes=['/health', '/users'])
True

"""Example 46: pytest verification for Declarative Config vs Setup."""
 
from example import Server, build_via_declared_spec, build_via_imperative_setup
 
 
def test_both_construction_styles_produce_an_equal_object() -> None:
    imperative = build_via_imperative_setup()  # => step-by-step mutation
    declared = build_via_declared_spec({"host": "api.example.com", "port": 443, "routes": ["/health", "/users"]})
    assert imperative == declared  # => dataclass structural equality
 
 
def test_declared_spec_with_different_values_builds_a_different_server() -> None:
    server = build_via_declared_spec({"host": "other.example.com", "port": 80, "routes": []})
    assert server == Server(host="other.example.com", port=80, routes=[])  # => matches the given spec exactly
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 47: Relational vs Nested-Loop Join."""
 
import sqlite3  # => the standard library's built-in SQL engine -- no external dependency needed
 
customers: list[tuple[int, str]] = [(1, "alice"), (2, "bob")]  # => (customer_id, name)
orders: list[tuple[int, int, str]] = [(101, 1, "widget"), (102, 2, "gadget"), (103, 1, "gizmo")]
# => (order_id, customer_id, item)
 
 
def join_via_sql(customers: list[tuple[int, str]], orders: list[tuple[int, int, str]]) -> list[tuple[str, str]]:  # => declarative leg
    conn = sqlite3.connect(":memory:")  # => declare tables, state the join, let SQLite compute it
    conn.execute("CREATE TABLE customers (id INTEGER, name TEXT)")  # => declares the shape, no data yet
    conn.execute("CREATE TABLE orders (id INTEGER, customer_id INTEGER, item TEXT)")  # => same, for orders
    conn.executemany("INSERT INTO customers VALUES (?, ?)", customers)  # => load every customer row
    conn.executemany("INSERT INTO orders VALUES (?, ?, ?)", orders)  # => load every order row
    rows = conn.execute(  # => the query IS the join -- no accumulator variable anywhere in this function
        "SELECT customers.name, orders.item FROM customers JOIN orders ON customers.id = orders.customer_id ORDER BY orders.id"
        # => JOIN...ON declares the relationship -- no explicit loop nesting anywhere in this code
    ).fetchall()  # => the query planner decided HOW to match rows; this call only asks for the results
    conn.close()  # => release the in-memory connection
    return rows  # => list of (name, item) pairs
 
 
def join_via_nested_loop(customers: list[tuple[int, str]], orders: list[tuple[int, int, str]]) -> list[tuple[str, str]]:  # => imperative leg
    result: list[tuple[str, str]] = []  # => mutable accumulator
    for _order_id, customer_id, item in orders:  # => outer loop: every order, in insertion order
        for cid, name in customers:  # => inner loop: scan every customer looking for a match
            if cid == customer_id:  # => the join condition, written out explicitly as a comparison
                result.append((name, item))  # => explicit accumulation, one matched pair at a time
                break  # => stop scanning customers once this order's match is found
    return result  # => the fully built accumulator
 
 
sql_result = join_via_sql(customers, orders)  # => declarative version
loop_result = join_via_nested_loop(customers, orders)  # => imperative version
 
print(sql_result)  # => both must produce identical (name, item) pairs, in the same order
# => Output: [('alice', 'widget'), ('bob', 'gadget'), ('alice', 'gizmo')]
print(sql_result == loop_result)  # => confirms the declarative and imperative joins agree
# => Output: True

python3 example.py

[('alice', 'widget'), ('bob', 'gadget'), ('alice', 'gizmo')]
True

"""Example 47: pytest verification for Relational vs Nested-Loop Join."""
 
from example import join_via_nested_loop, join_via_sql
 
 
def test_sql_and_nested_loop_joins_produce_identical_rows() -> None:
    customers = [(1, "alice"), (2, "bob")]  # => same fixtures as the module-level demo
    orders = [(101, 1, "widget"), (102, 2, "gadget"), (103, 1, "gizmo")]
    assert join_via_sql(customers, orders) == join_via_nested_loop(customers, orders)
 
 
def test_an_order_with_no_matching_customer_is_dropped_by_both_joins() -> None:
    customers = [(1, "alice")]  # => only customer 1 exists
    orders = [(101, 1, "widget"), (102, 99, "orphan")]  # => order 102 references a nonexistent customer
    sql_rows = join_via_sql(customers, orders)  # => an INNER JOIN drops unmatched orders
    loop_rows = join_via_nested_loop(customers, orders)  # => the nested loop's `break` also drops it
    assert sql_rows == loop_rows == [("alice", "widget")]  # => both agree: the orphan order vanishes
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 48: Pure Core, Imperative Shell."""
 
 
def compute_invoice_total(unit_price: int, quantity: int, discount_pct: int) -> int:  # => the PURE CORE
    subtotal = unit_price * quantity  # => no I/O, no globals, only its own arguments
    discount = subtotal * discount_pct // 100  # => pure arithmetic
    return subtotal - discount  # => deterministic: same inputs always produce the same output
    # => this function can be tested with zero I/O, zero mocks, zero setup -- that is the whole point
 
 
def print_invoice(unit_price: int, quantity: int, discount_pct: int) -> None:  # => the IMPERATIVE SHELL
    total = compute_invoice_total(unit_price, quantity, discount_pct)  # => delegate all logic to the core
    print(f"Total: {total}")  # => the ONLY line in this file that performs I/O -- the shell's whole job
    # => the shell contains no business logic of its own -- it just wires the pure core to the outside world
 
 
print_invoice(1000, 3, 10)  # => 1000*3=3000, 10% off = 300, total 2700
# => Output: Total: 2700

python3 example.py

Total: 2700

"""Example 48: pytest verification for Pure Core, Imperative Shell."""
 
from example import compute_invoice_total
 
 
def test_core_is_tested_with_zero_io_and_zero_mocks() -> None:
    # => this test never touches print(), a file, or a network call -- proves the core needs no I/O
    assert compute_invoice_total(1000, 3, 10) == 2700  # => 3000 - 300
 
 
def test_core_is_deterministic_across_repeated_calls() -> None:
    first = compute_invoice_total(500, 2, 20)  # => call #1
    second = compute_invoice_total(500, 2, 20)  # => call #2, identical arguments
    assert first == second == 800  # => 1000 - 200, same both times -- no hidden state anywhere
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 49: Multi-Paradigm Boundary."""
 
from dataclasses import dataclass
 
 
def functional_pipeline(raw_prices: tuple[int, ...]) -> tuple[int, ...]:  # => a pure functional pipeline
    discounted = tuple(p - (p * 10 // 100) for p in raw_prices)  # => map: apply a 10% discount
    return tuple(p for p in discounted if p > 0)  # => filter: drop non-positive prices
    # => every step returns a NEW immutable tuple -- nothing here is ever mutated in place
 
 
@dataclass  # => an OO service the pipeline hands its result to, across a clean boundary
class InventoryService:
    accepted_prices: list[int]
 
    def record_batch(self, prices: tuple[int, ...]) -> None:  # => the ONLY place mutation happens
        self.accepted_prices.extend(prices)  # => OO-style in-place mutation, but confined to this class
 
 
raw_prices = (100, 50, 5, 200)  # => an immutable tuple -- the functional side's input
cleaned = functional_pipeline(raw_prices)  # => pure functional processing, no mutation anywhere yet
print(cleaned)  # => 100->90, 50->45, 5->5 (5*10//100=0 discount, still positive), 200->180; nothing dropped
# => Output: (90, 45, 5, 180)
 
service = InventoryService(accepted_prices=[])  # => the OO side of the boundary
service.record_batch(cleaned)  # => the boundary: an immutable tuple crosses into a mutable OO object
print(service.accepted_prices)  # => confirms the OO side received exactly the functional side's output
# => Output: [90, 45, 5, 180]
print(cleaned)  # => the tuple itself is STILL untouched -- crossing the boundary never mutated it
# => Output: (90, 45, 5, 180)

python3 example.py

(90, 45, 5, 180)
[90, 45, 5, 180]
(90, 45, 5, 180)

"""Example 49: pytest verification for Multi-Paradigm Boundary."""
 
from example import InventoryService, functional_pipeline
 
 
def test_boundary_passes_only_immutable_data() -> None:
    cleaned = functional_pipeline((100, 50, 5, 200))  # => run the pure side
    assert isinstance(cleaned, tuple)  # => the boundary value itself is immutable
 
    service = InventoryService(accepted_prices=[])  # => fresh OO service
    service.record_batch(cleaned)  # => cross the boundary
    assert service.accepted_prices == [90, 45, 5, 180]  # => the OO side received the pipeline's output
    assert cleaned == (90, 45, 5, 180)  # => and the tuple itself is provably unchanged after crossing
 
 
def test_functional_side_never_mutates_its_own_input_tuple() -> None:
    raw = (10, 20)  # => a small immutable input
    functional_pipeline(raw)  # => call once, discard the result -- only checking for mutation
    assert raw == (10, 20)  # => tuples cannot be mutated in place at all, but this documents the contract
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 50: Paradigm Soup Anti-Pattern."""
 
 
class MutableBucket:  # => an OO object with mutable state, threaded through a nominally "functional" pipeline
    def __init__(self, items: list[int]) -> None:
        self.items = items  # => a MUTABLE list, not a tuple -- this is the seed of the bug
 
 
def add_bonus_functional_looking(bucket: MutableBucket) -> MutableBucket:  # => LOOKS like a pure map step...
    bucket.items.append(999)  # => ...but secretly MUTATES the shared list in place -- paradigm soup
    return bucket  # => returning the SAME mutated object, not a new one, is the tell
 
 
def scale_functional_looking(bucket: MutableBucket) -> MutableBucket:  # => a second "map step"
    for i in range(len(bucket.items)):  # => also mutates in place, hidden behind a function-call facade
        bucket.items[i] *= 2  # => in-place scaling
    return bucket  # => same object identity as the input -- no new value was actually created
 
 
original = MutableBucket([1, 2, 3])  # => construct the shared mutable object once
step1 = add_bonus_functional_looking(original)  # => "looks like" pipe(original, add_bonus)
step2 = scale_functional_looking(step1)  # => "looks like" pipe(step1, scale) -- chained, functional style
 
print(step2.items)  # => the visible chained result
# => Output: [2, 4, 6, 1998]
print(original.items)  # => THE BUG: `original` was aliased and mutated by every "pipeline" step
# => Output: [2, 4, 6, 1998]
print(original is step1 is step2)  # => all three names point at the SAME object -- no new values anywhere
# => Output: True

python3 example.py

[2, 4, 6, 1998]
[2, 4, 6, 1998]
True

"""Example 50: pytest verification for Paradigm Soup Anti-Pattern."""
 
from example import MutableBucket, add_bonus_functional_looking, scale_functional_looking
 
 
def test_aliasing_bug_reproduces_original_is_silently_mutated() -> None:
    original = MutableBucket([1, 2, 3])  # => fresh bucket, isolated from the module-level demo
    result = scale_functional_looking(add_bonus_functional_looking(original))  # => a "functional-looking" chain
 
    assert result.items == [2, 4, 6, 1998]  # => the chained result
    assert original.items == [2, 4, 6, 1998]  # => THE BUG: the "original" reference was mutated too
    assert original is result  # => same object identity -- no new value was ever produced
 
 
def test_a_true_immutable_pipeline_would_not_have_this_bug() -> None:
    original: tuple[int, ...] = (1, 2, 3)  # => the honest functional fix: use an immutable tuple instead
    bonus = original + (999,)  # => a genuinely NEW tuple, original untouched
    scaled = tuple(n * 2 for n in bonus)  # => another genuinely NEW tuple
    assert original == (1, 2, 3)  # => the true-functional version never mutates the original at all
    assert scaled == (2, 4, 6, 1998)  # => and still produces the identical final values
 
 
# => Run: pytest -- Output: 2 passed

pytest -q

2 passed

"""Example 51: Logic vs Imperative Reachability."""
 
from collections import deque  # => deque gives O(1) popleft(), the BFS queue's core operation
 
edges: dict[str, list[str]] = {  # => a directed graph, shared by both approaches below
    "a": ["b", "c"],  # => a points to b and c
    "b": ["d"],  # => b points to d
    "c": [],  # => c has no outgoing edges
    "d": ["a"],  # => a cycle back to "a" -- both approaches must handle this without looping forever
    "e": [],  # => an unreachable, isolated node
}  # => closes the shared graph declaration
 
 
def reachable_via_inference(start: str, edges: dict[str, list[str]]) -> set[str]:  # => LOGIC-flavored
    # => rule: reachable(X, Y) :- edge(X, Y).  reachable(X, Z) :- edge(X, Y), reachable(Y, Z).
    known: set[str] = set()  # => the set of facts inferred so far (a fixed-point computation)
    frontier = set(edges.get(start, []))  # => seed with everything directly reachable via one edge
    while frontier - known:  # => keep inferring new facts until nothing new can be derived (fixed point)
        newly_known = frontier - known  # => facts inferred in THIS round that weren't already known
        known |= newly_known  # => add them to the known set
        frontier = known | {y for x in newly_known for y in edges.get(x, [])}  # => derive one more hop
    return known  # => the full set of inferred "reachable" facts
 
 
def reachable_via_bfs(start: str, edges: dict[str, list[str]]) -> set[str]:  # => IMPERATIVE: explicit BFS
    visited: set[str] = set()  # => mutable set, built up by explicit traversal
    queue: deque[str] = deque(edges.get(start, []))  # => explicit FIFO work queue
    while queue:  # => explicit loop draining the queue
        node = queue.popleft()  # => explicit dequeue
        if node in visited:  # => explicit cycle guard
            continue  # => skip re-processing an already-visited node -- prevents the cycle from looping forever
        visited.add(node)  # => explicit mutation
        queue.extend(edges.get(node, []))  # => explicit enqueue of newly discovered neighbors
    return visited  # => the fully built set
 
 
inference_result = reachable_via_inference("a", edges)  # => run the logic-flavored version
bfs_result = reachable_via_bfs("a", edges)  # => run the imperative version
 
print(sorted(inference_result))  # => a -> b,c; b -> d; d -> a: the cycle makes "a" reachable from itself too
# => Output: ['a', 'b', 'c', 'd']
print(inference_result == bfs_result)  # => both must compute the identical reachable set
# => Output: True