Intermediate Examples

"""Example 29: Extract a Pure Core from a Mutating Routine."""
 
report_log: list[str] = []  # => the ORIGINAL routine's hidden side effect target
 
 
def process_orders_impure(orders: list[int]) -> int:  # => mixes computation AND I/O together
    total = 0  # => the running sum, mutated below
    for amount in orders:  # => a loop mutating both total and the module-level log
        total += amount  # => mutates total AND appends below -- two side effects per iteration
        report_log.append(f"processed {amount}")  # => side effect buried inside the loop
    return total  # => the caller cannot test this without also inspecting report_log
 
 
def sum_orders_pure(orders: list[int]) -> int:  # => the EXTRACTED core -- no I/O, no logging
    return sum(orders)  # => a pure fold, trivially testable with no mocking
 
 
def make_log_lines(orders: list[int]) -> list[str]:  # => the logging, extracted as its OWN pure function
    return [f"processed {amount}" for amount in orders]  # => still pure: builds and returns, doesn't print
 
 
orders = [10, 20, 30]  # => shared input for both the impure routine and the pure core
 
impure_total = process_orders_impure(orders)  # => also mutates report_log as a side effect
pure_total = sum_orders_pure(orders)  # => computes the SAME total with zero side effects
 
# => this is the co-28 functional-core/imperative-shell split in miniature
print(impure_total == pure_total)  # => Output: True -- same answer, radically different testability
print(make_log_lines(orders) == report_log)  # => Output: True -- logging extracted without losing it

True
True

"""Example 29: pytest verification for Extract a Pure Core from a Mutating Routine."""
 
from example import make_log_lines, sum_orders_pure
 
 
def test_pure_core_needs_no_mocking() -> None:
    orders = [1, 2, 3]
    assert sum_orders_pure(orders) == 6  # => tested with plain arguments, no I/O to intercept
 
 
def test_log_lines_extracted_as_pure_function() -> None:
    orders = [5, 15]
    assert make_log_lines(orders) == ["processed 5", "processed 15"]
 
 
# => Run: pytest -- Output: 2 passed

2 passed

"""Example 30: O(1) Sharing on a Persistent Linked List."""
 
from __future__ import annotations  # => enables the quoted 'PList | None' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds the immutable PList node
 
 
@dataclass(frozen=True)  # => a persistent list node caching its own length -- O(1) to read
class PList:  # => the node type itself -- see the decorator above for immutability
    head: int  # => this node's own value
    tail: "PList | None"  # => the shared, untouched rest of the list
    length: int  # => cached at construction time -- never recomputed by walking
 
 
def plist_prepend(value: int, lst: "PList | None") -> PList:  # => O(1): no walk over lst needed
    previous_length = lst.length if lst is not None else 0  # => O(1) read of a cached field
    return PList(head=value, tail=lst, length=previous_length + 1)  # => reuses lst AS-IS
 
 
empty: PList | None = None  # => the shared empty base every version points back to
version_a = plist_prepend(1, empty)  # => version_a.length is 1
version_b = plist_prepend(2, version_a)  # => version_b.length is 2, REUSING version_a untouched
 
# => structural sharing means version_b costs O(1) extra memory, not O(n)
print(version_a.length)  # => Output: 1
print(version_b.length)  # => Output: 2
print(version_b.tail is version_a)  # => Output: True -- structural sharing, not a copy
print(version_a.head)  # => Output: 1 -- version_a itself was never touched by building version_b

1
2
True
1

"""Example 30: pytest verification for O(1) Sharing on a Persistent Linked List."""
 
from example import PList, plist_prepend
 
 
def test_prepend_is_o1_and_shares_the_tail() -> None:
    version_a: PList | None = plist_prepend(1, None)
    version_b = plist_prepend(2, version_a)  # => O(1): only reads version_a.length, never walks it
 
    assert version_b.length == 2
    assert version_a.length == 1  # => unaffected by building version_b
    assert version_b.tail is version_a  # => structural sharing, not a copy
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 31: A Read-Only View via MappingProxyType."""
 
from types import MappingProxyType  # => a read-only VIEW over an existing dict, not a copy
 
config = {"retries": 3, "timeout": 5}  # => the underlying mutable dict this example protects
readonly_config = MappingProxyType(config)  # => wraps config -- reads pass through, writes are blocked
# => this is the co-04 "immutability at the boundary" pattern -- share config without risking mutation
 
print(readonly_config["retries"])  # => Output: 3 -- reads work exactly like a normal dict
 
try:  # => opens a block that expects the write below to raise
    readonly_config["retries"] = 99  # type: ignore[index]  # => attempts a write through the view
    raised = False  # => unreachable if TypeError fires
except TypeError:  # => MappingProxyType has no __setitem__
    raised = True  # => confirms the view actually blocked the write
 
print(raised)  # => Output: True
config["retries"] = 10  # => the UNDERLYING dict itself is still mutable directly
print(readonly_config["retries"])  # => Output: 10 -- the view reflects the underlying dict LIVE

3
True
10

"""Example 31: pytest verification for A Read-Only View via MappingProxyType."""
 
from types import MappingProxyType
 
 
def test_view_blocks_writes_but_reflects_the_underlying_dict() -> None:
    config = {"retries": 3}
    readonly_config = MappingProxyType(config)
    try:
        readonly_config["retries"] = 99  # type: ignore[index]
        raised = False
    except TypeError:
        raised = True
    assert raised is True
 
    config["retries"] = 10  # => mutate the underlying dict directly
    assert readonly_config["retries"] == 10  # => the view is LIVE, not a snapshot
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 32: Stateful Closure Counter vs. Pure Fold."""
 
from functools import reduce  # => reduce is used by the pure fold below
from typing import Callable  # => Callable types the closure returned by make_counter
 
 
def make_counter() -> Callable[[], int]:  # => returns a closure holding MUTABLE state
    count = 0  # => lives INSIDE the closure -- state lives here, nowhere else
 
    def increment() -> int:  # => the returned closure itself
        nonlocal count  # => declares intent to mutate the ENCLOSING count, not a local copy
        count += 1  # => side effect: mutates state captured by the closure
        return count  # => returns the count AFTER incrementing
 
    return increment  # => make_counter itself returns the closure, not a value
 
 
def count_pure(n: int) -> int:  # => a pure fold: no closure, no mutation, no hidden state
    return reduce(lambda acc, _: acc + 1, range(n), 0)  # => the SAME total, computed with zero state
 
 
counter = make_counter()  # => a fresh closure with its own private count
stateful_result = [counter(), counter(), counter()]  # => each call MUTATES the shared closure state
pure_result = count_pure(3)  # => a single expression, no calls needed, no state anywhere
 
# => co-08 closures vs co-01 purity: same answer, two different state models
print(stateful_result)  # => Output: [1, 2, 3]
print(pure_result)  # => Output: 3
print(stateful_result[-1] == pure_result)  # => Output: True -- same final count, different state model

[1, 2, 3]
3
True

"""Example 32: pytest verification for Stateful Closure Counter vs. Pure Fold."""
 
from example import count_pure, make_counter
 
 
def test_closure_state_and_pure_fold_agree_on_the_final_count() -> None:
    counter = make_counter()
    for _ in range(3):
        counter()
    assert counter() == 4  # => state lives IN the closure, mutated across calls
    assert count_pure(4) == 4  # => the SAME count, computed with zero mutable state
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 33: A Pipeline of partial Calls."""
 
from functools import partial  # => partial is the currying mechanism used below
 
 
def scale(factor: int, x: int) -> int:  # => a 2-argument function -- factor fixed, x supplied later
    return factor * x  # => the actual multiplication scale performs
 
 
def offset(amount: int, x: int) -> int:  # => a second 2-argument function, same shape
    return amount + x  # => the actual addition offset performs
 
 
double = partial(scale, 2)  # => fixes factor=2 -- a reusable "double" function
add_ten = partial(offset, 10)  # => fixes amount=10 -- a reusable "add ten" function
 
pipeline_result = add_ten(double(5))  # => double(5) is 10, then add_ten(10) is 20
manual_result = 10 + (2 * 5)  # => the equivalent hand-written arithmetic, for comparison
 
# => partial is Python's practical stand-in for true currying (co-10)
print(pipeline_result)  # => Output: 20
print(pipeline_result == manual_result)  # => Output: True -- partials compose like ordinary functions
print(double(100))  # => Output: 200 -- double is REUSABLE across any input, not just 5

20
True
200

"""Example 33: pytest verification for A Pipeline of partial Calls."""
 
from functools import partial
 
from example import offset, scale
 
 
def test_chained_partials_compose_like_ordinary_functions() -> None:
    double = partial(scale, 2)
    add_ten = partial(offset, 10)
    assert add_ten(double(5)) == 20
    assert double(100) == 200  # => double is reusable across any input
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 34: compose(*fns) Folds a List of Functions."""
 
from functools import reduce  # => reduce folds the chain of functions inside compose
from typing import Callable  # => Callable types every function compose accepts
 
 
def compose(*fns: Callable[[int], int]) -> Callable[[int], int]:  # => any NUMBER of functions
    def composed(x: int) -> int:  # => the returned pipeline function
        return reduce(lambda acc, fn: fn(acc), reversed(fns), x)  # => rightmost fn runs FIRST
        # => reversed(fns) makes compose(f, g, h)(x) mean f(g(h(x))), matching math notation
 
    return composed  # => compose itself returns the pipeline function, not a value
 
 
def add_one(x: int) -> int:  # => the innermost step in the pipeline below
    return x + 1  # => the actual +1 add_one performs
 
 
def double(x: int) -> int:  # => the middle step in the pipeline below
    return x * 2  # => the actual *2 double performs
 
 
def square(x: int) -> int:  # => the outermost step in the pipeline below
    return x * x  # => the actual squaring square performs
 
 
pipeline = compose(square, double, add_one)  # => reads right-to-left: square(double(add_one(x)))
result = pipeline(3)  # => add_one(3)=4, double(4)=8, square(8)=64
 
print(result)  # => Output: 64
print(result == square(double(add_one(3))))  # => Output: True -- compose(*fns) matches nested calls

64
True

"""Example 34: pytest verification for compose(*fns) Folds a List of Functions."""
 
from example import add_one, compose, double, square
 
 
def test_compose_star_matches_nested_application_order() -> None:
    pipeline = compose(square, double, add_one)
    assert pipeline(3) == square(double(add_one(3))) == 64
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 35: Rewriting a Lambda Pipeline Point-Free."""
 
from functools import partial, reduce  # => partial curries add/multiply; reduce powers compose
from typing import Callable  # => Callable types the pipeline function returned by compose
 
 
def compose(*fns: Callable[[int], int]) -> Callable[[int], int]:  # => folds fns right-to-left
    def composed(x: int) -> int:  # => the returned pipeline function
        return reduce(lambda acc, fn: fn(acc), reversed(fns), x)  # => rightmost fn runs FIRST, matching math notation
 
    return composed  # => compose itself returns the pipeline function
 
 
def add(n: int, x: int) -> int:  # => a 2-argument helper, curried via partial below
    return n + x  # => the actual addition add performs
 
 
def multiply(n: int, x: int) -> int:  # => a second 2-argument helper, same shape
    return n * x  # => the actual multiplication multiply performs
 
 
with_named_arg: Callable[[int], int] = lambda x: multiply(2, add(3, x))  # => POINTED: names x explicitly
point_free = compose(partial(multiply, 2), partial(add, 3))  # => POINT-FREE: x never named
 
# => point-free style trades explicit naming for compact composition (co-19)
print(with_named_arg(5))  # => Output: 16  ((5 + 3) * 2)
print(point_free(5))  # => Output: 16 -- identical result, x never appears in point_free's own definition
print(with_named_arg(5) == point_free(5))  # => Output: True -- two styles, one computation

16
16
True

"""Example 35: pytest verification for Rewriting a Lambda Pipeline Point-Free."""
 
from functools import partial
from typing import Callable
 
from example import add, compose, multiply
 
 
def test_point_free_matches_the_named_argument_version() -> None:
    with_named_arg: Callable[[int], int] = lambda x: multiply(2, add(3, x))
    point_free = compose(partial(multiply, 2), partial(add, 3))
    assert point_free(5) == with_named_arg(5) == 16
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 36: Building a Histogram with reduce."""
 
from functools import reduce  # => reduce folds words into the histogram below
 
 
def add_to_histogram(  # => the "how to fold one more element in" step
    histogram: dict[str, int], word: str  # => the accumulator and the next element, reduce's own calling convention
) -> dict[str, int]:  # => closes the multi-line signature above
    histogram[word] = histogram.get(word, 0) + 1  # => increments, defaulting missing keys to 0
    return histogram  # => reduce expects the combiner to return the NEW accumulator
 
 
words = ["a", "b", "a", "c", "b", "a"]  # => the source sequence being counted
 
histogram = reduce(add_to_histogram, words, {})  # => folds words into ONE dict, seeded empty
 
# => reduce is the general-purpose fold every other aggregation specializes
print(histogram)  # => Output: {'a': 3, 'b': 2, 'c': 1}
print(histogram["a"])  # => Output: 3 -- 'a' appeared three times
print(sum(histogram.values()) == len(words))  # => Output: True -- every word counted exactly once

{'a': 3, 'b': 2, 'c': 1}
3
True

"""Example 36: pytest verification for Building a Histogram with reduce."""
 
from functools import reduce
 
from example import add_to_histogram
 
 
def test_reduce_builds_a_correct_histogram() -> None:
    words = ["x", "y", "x"]
    histogram = reduce(add_to_histogram, words, {})
    assert histogram == {"x": 2, "y": 1}
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 37: Chaining map, filter, and reduce."""
 
from functools import reduce  # => reduce folds the filtered stream into one total
 
orders = [12, 7, 25, 3, 18, 9]  # => raw order amounts, some too small to matter
 
doubled = map(lambda amount: amount * 2, orders)  # => LAZY: doubles every order (e.g. a 2x promo)
significant = filter(lambda amount: amount > 20, doubled)  # => LAZY: keeps only large-enough orders
total = reduce(lambda acc, amount: acc + amount, significant, 0)  # => folds survivors into one sum
# => each stage is lazy until reduce finally pulls every value through the whole pipeline
 
manual_total = sum(a * 2 for a in orders if a * 2 > 20)  # => the equivalent single comprehension
 
print(total)  # => Output: 110
print(total == manual_total)  # => Output: True -- three separate stages, one verified answer

110
True

"""Example 37: pytest verification for Chaining map, filter, and reduce."""
 
from functools import reduce
 
 
def test_pipeline_matches_the_equivalent_comprehension() -> None:
    orders = [12, 7, 25, 3, 18, 9]
    doubled = map(lambda amount: amount * 2, orders)
    significant = filter(lambda amount: amount > 20, doubled)
    total = reduce(lambda acc, amount: acc + amount, significant, 0)
    assert total == sum(a * 2 for a in orders if a * 2 > 20) == 110
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 38: A Lazy map/filter Pipeline Pulled by next."""
 
from typing import Iterator  # => Iterator types the lazy generator below
 
pulled_from_source: list[int] = []  # => tracks exactly which SOURCE values were ever touched
 
 
def logged_source() -> Iterator[int]:  # => a generator that records every value it produces
    for n in range(1, 1000):  # => a LARGE range -- but nothing runs until pulled
        pulled_from_source.append(n)  # => only executes when this generator is actually advanced
        yield n  # => suspends here until the consumer pulls the NEXT value
 
 
doubled = map(lambda n: n * 2, logged_source())  # => LAZY: wraps the source, still nothing runs
evens_only = filter(lambda n: n % 4 == 0, doubled)  # => LAZY: a second lazy stage on top of the first
 
first_three = [next(evens_only), next(evens_only), next(evens_only)]  # => pulls JUST enough
 
# => laziness (co-15) means the pipeline computes ONLY what the caller demands
print(first_three)  # => Output: [4, 8, 12]
print(len(pulled_from_source) < 10)  # => Output: True -- far fewer than 999 source values were touched

[4, 8, 12]
True

"""Example 38: pytest verification for A Lazy map/filter Pipeline Pulled by next."""
 
from typing import Iterator
 
 
def test_pipeline_pulls_only_as_many_source_values_as_needed() -> None:
    pulled: list[int] = []
 
    def source() -> Iterator[int]:
        for n in range(1, 1000):
            pulled.append(n)
            yield n
 
    doubled = map(lambda n: n * 2, source())
    evens_only = filter(lambda n: n % 4 == 0, doubled)
    result = [next(evens_only), next(evens_only)]
    assert result == [4, 8]
    assert len(pulled) < 10  # => the pipeline never touched most of the 999-element range
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 39: Running Totals via accumulate."""
 
from itertools import accumulate  # => a lazy iterator of running totals (or any binary op)
 
deposits = [100, 50, -30, 200]  # => a sequence of account movements
 
running_balance = list(accumulate(deposits))  # => default op is +: each step is the sum SO FAR
running_max = list(accumulate(deposits, max))  # => a custom op: running maximum instead of sum
# => accumulate is itertools' generalized fold-that-keeps-every-intermediate-result
 
print(running_balance)  # => Output: [100, 150, 120, 320]
print(running_max)  # => Output: [100, 100, 100, 200]
print(running_balance[-1] == sum(deposits))  # => Output: True -- the LAST running total is the full sum

[100, 150, 120, 320]
[100, 100, 100, 200]
True

"""Example 39: pytest verification for Running Totals via accumulate."""
 
from itertools import accumulate
 
 
def test_accumulate_produces_prefix_sums() -> None:
    deposits = [10, 20, 30]
    assert list(accumulate(deposits)) == [10, 30, 60]
    assert list(accumulate(deposits, max)) == [10, 20, 30]
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 40: Adjacent Pairs via pairwise and tee."""
 
from itertools import pairwise, tee  # => pairwise: consecutive pairs; tee: split one iterator in two
 
readings = [10, 12, 9, 15, 15]  # => a sequence of sensor readings
 
deltas = [b - a for a, b in pairwise(readings)]  # => consecutive differences, one per adjacent pair
# => pairwise([10, 12, 9, 15, 15]) yields (10,12), (12,9), (9,15), (15,15) -- 4 pairs from 5 readings
 
original_stream, backup_stream = tee(iter(readings))  # => splits ONE iterator into two INDEPENDENT ones
first_from_original = next(original_stream)  # => advances original_stream only
first_from_backup = next(backup_stream)  # => backup_stream is UNAFFECTED by the pull above
 
print(deltas)  # => Output: [2, -3, 6, 0]
print(first_from_original == first_from_backup)  # => Output: True -- both streams start from readings[0]

[2, -3, 6, 0]
True

"""Example 40: pytest verification for Adjacent Pairs via pairwise and tee."""
 
from itertools import pairwise, tee
 
 
def test_pairwise_and_tee() -> None:
    readings = [1, 2, 4]
    assert list(pairwise(readings)) == [(1, 2), (2, 4)]
 
    stream_a, stream_b = tee(iter(readings))
    assert next(stream_a) == next(stream_b) == 1  # => independent streams, same starting point
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 41: A Hand-Rolled Memoization Dict."""
 
from typing import Callable  # => Callable types the function memoize wraps
 
call_count = 0  # => tracks how many times the EXPENSIVE function body actually ran
 
 
def expensive(n: int) -> int:  # => stands in for a slow, pure computation
    global call_count  # => declares intent to mutate the MODULE-level counter
    call_count += 1  # => only increments when the body actually executes
    return n * n  # => the actual (slow, pure) computation being cached
 
 
def memoize(fn: Callable[[int], int]) -> Callable[[int], int]:  # => a hand-rolled cache, no lru_cache
    cache: dict[int, int] = {}  # => the memo table, one entry per unique argument
 
    def wrapper(n: int) -> int:  # => the returned, cache-checking function
        if n not in cache:  # => cache MISS: compute once and store
            cache[n] = fn(n)  # => computes ONCE and stores under key n
        return cache[n]  # => cache HIT or freshly-stored value, either way no recomputation
 
    return wrapper  # => memoize itself returns the wrapping function
 
 
memoized_expensive = memoize(expensive)  # => wraps expensive with the hand-rolled cache
 
first = memoized_expensive(5)  # => cache miss -- call_count becomes 1
second = memoized_expensive(5)  # => cache HIT -- call_count stays 1
third = memoized_expensive(6)  # => a NEW argument -- cache miss, call_count becomes 2
 
# => memoization trades memory for time, valid ONLY because expensive is pure
print(first, second, third)  # => Output: 25 25 36
print(call_count)  # => Output: 2 -- expensive body ran exactly twice, for the two unique arguments

25 25 36
2

"""Example 41: pytest verification for A Hand-Rolled Memoization Dict."""
 
from example import memoize
 
 
def test_second_call_is_served_from_cache() -> None:
    calls: list[int] = []
 
    def track(n: int) -> int:
        calls.append(n)
        return n * n
 
    memoized = memoize(track)
    assert memoized(4) == 16
    assert memoized(4) == 16
    assert calls == [4]  # => the second call never re-ran the underlying function
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 42: A Parameterized @retry(3) Decorator."""
 
from typing import Callable  # => Callable types every layer of this decorator factory
 
attempt_log: list[int] = []  # => records which attempt number ran, for verification below
 
 
def retry(times: int) -> Callable[[Callable[[], str]], Callable[[], str]]:  # => a decorator FACTORY
    def decorator(fn: Callable[[], str]) -> Callable[[], str]:  # => the ACTUAL decorator, closes over times
        def wrapper() -> str:  # => the actual retry loop lives here
            last_error: Exception | None = None  # => remembers the most recent failure, if any
            for attempt in range(1, times + 1):  # => tries up to `times` times, closed over from retry()
                attempt_log.append(attempt)  # => logs every attempt, successful or not
                try:  # => attempts ONE call to the wrapped function
                    return fn()  # => success: return immediately, no further attempts
                except ValueError as exc:  # => a failed attempt -- remember it and try again
                    last_error = exc  # => remembers the failure so it can be re-raised later
            raise last_error  # type: ignore[misc]  # => every attempt failed -- re-raise the last error
 
        return wrapper  # => decorator itself returns the retry-wrapped function
 
    return decorator  # => retry(times) itself returns the decorator
 
 
calls_before_success = 0  # => simulates a flaky operation succeeding on its 3rd attempt
 
 
@retry(3)  # => equivalent to: flaky = retry(3)(flaky) -- retry(3) runs FIRST, returns a decorator
def flaky() -> str:  # => the function actually being retried
    global calls_before_success  # => declares intent to mutate the MODULE-level counter
    calls_before_success += 1  # => tracks how many times this call has been attempted
    if calls_before_success < 3:  # => fails on attempts 1 and 2
        raise ValueError("not yet")  # => simulates a transient failure on early attempts
    return "ok"  # => succeeds on attempt 3
 
 
result = flaky()  # => retries internally until success, or until times is exhausted
 
# => a decorator factory is a function that RETURNS a decorator, one extra layer
print(result)  # => Output: ok
print(attempt_log)  # => Output: [1, 2, 3] -- three attempts were logged before success

ok
[1, 2, 3]

"""Example 42: pytest verification for A Parameterized @retry(3) Decorator."""
 
from example import retry
 
 
def test_retry_stops_as_soon_as_the_wrapped_call_succeeds() -> None:
    attempts: list[int] = []
 
    @retry(5)
    def sometimes_fails() -> str:
        attempts.append(len(attempts) + 1)
        if len(attempts) < 2:
            raise ValueError("not yet")
        return "ok"
 
    assert sometimes_fails() == "ok"
    assert attempts == [1, 2]  # => stopped retrying the moment it succeeded
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 43: functools.wraps Preserves __name__."""
 
from functools import wraps  # => wraps is the fix demonstrated in careful_decorator
from typing import Callable  # => Callable types both decorators below
 
 
def naive_decorator(fn: Callable[[], str]) -> Callable[[], str]:  # => WITHOUT functools.wraps
    def wrapper() -> str:  # => the naive wrapper -- no metadata copied
        return fn()  # => forwards the call, nothing else
 
    return wrapper  # => wrapper has its OWN __name__, "wrapper" -- fn's identity is lost
 
 
def careful_decorator(fn: Callable[[], str]) -> Callable[[], str]:  # => WITH functools.wraps
    @wraps(fn)  # => copies __name__, __doc__, and more from fn onto wrapper
    def wrapper() -> str:  # => the careful wrapper -- decorated with @wraps below
        return fn()  # => forwards the call, identical behavior to the naive version
 
    return wrapper  # => wrapper now REPORTS as if it were fn itself
 
 
@naive_decorator  # => applies the WITHOUT-wraps decorator
def greet_naive() -> str:  # => the function whose identity gets lost
    """Say hello, naively decorated."""  # => this docstring is LOST from greet_naive.__doc__ after decoration
    return "hello"  # => the actual greeting
 
 
@careful_decorator  # => applies the WITH-wraps decorator
def greet_careful() -> str:  # => the function whose identity survives decoration
    """Say hello, carefully decorated."""  # => this docstring IS preserved on greet_careful.__doc__
    return "hello"  # => the actual greeting
 
 
# => functools.wraps matters for debugging, introspection, and framework compatibility
print(greet_naive.__name__)  # => Output: wrapper -- identity LOST, unhelpful in tracebacks/introspection
print(greet_careful.__name__)  # => Output: greet_careful -- identity PRESERVED by functools.wraps
print(greet_careful.__doc__)  # => Output: Say hello, carefully decorated.

wrapper
greet_careful
Say hello, carefully decorated.

"""Example 43: pytest verification for functools.wraps Preserves __name__."""
 
from example import careful_decorator, naive_decorator
 
 
def test_wraps_preserves_name_and_plain_decorator_does_not() -> None:
    @naive_decorator
    def naive() -> str:
        return "x"
 
    @careful_decorator
    def careful() -> str:
        return "x"
 
    assert naive.__name__ == "wrapper"  # => identity lost without functools.wraps
    assert careful.__name__ == "careful"  # => identity preserved with functools.wraps
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 44: Converting Deep Recursion to an Explicit Stack."""
 
import sys  # => sys.getrecursionlimit reveals the current recursion ceiling
 
 
def sum_to_n_recursive(n: int) -> int:  # => the natural recursive definition -- one frame per call
    if n == 0:  # => base case: nothing left to add
        return 0  # => the base case's value
    return n + sum_to_n_recursive(n - 1)  # => grows the call stack by exactly one frame per call
 
 
def sum_to_n_iterative(n: int) -> int:  # => the SAME computation, an explicit stack instead of the call stack
    stack: list[int] = list(range(1, n + 1))  # => an explicit, heap-allocated stack -- NOT the call stack
    total = 0  # => the running total, mutated below
    while stack:  # => pops until the explicit stack is empty -- no Python call frames grow
        total += stack.pop()  # => pops one value at a time -- heap memory, not call-stack frames
    return total  # => the final total after the explicit stack empties
 
 
deep_n = sys.getrecursionlimit() + 500  # => deliberately EXCEEDS the recursion ceiling
 
try:  # => opens a block that expects the recursive call below to raise
    sum_to_n_recursive(deep_n)  # => this call is expected to blow the call stack
    recursive_raised = False  # => unreachable if RecursionError fires as expected
except RecursionError:  # => confirms the exact failure recursion has at this depth
    recursive_raised = True  # => confirms the recursive version genuinely failed here
 
iterative_result = sum_to_n_iterative(deep_n)  # => the SAME depth, but no call-stack growth at all
 
# => CPython has NO tail-call optimization -- deep recursion genuinely can crash
print(recursive_raised)  # => Output: True -- the recursive version genuinely fails at this depth
print(iterative_result == deep_n * (deep_n + 1) // 2)  # => Output: True -- the iterative version succeeds

True
True

"""Example 44: pytest verification for Converting Deep Recursion to an Explicit Stack."""
 
import sys
 
from example import sum_to_n_iterative, sum_to_n_recursive
 
 
def test_iterative_version_survives_a_depth_that_breaks_recursion() -> None:
    deep_n = sys.getrecursionlimit() + 500
    try:
        sum_to_n_recursive(deep_n)
        recursive_raised = False
    except RecursionError:
        recursive_raised = True
    assert recursive_raised is True
 
    assert sum_to_n_iterative(deep_n) == deep_n * (deep_n + 1) // 2
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 45: A Shape ADT as a Union of Frozen Dataclasses."""
 
from __future__ import annotations  # => enables the quoted forward references used below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds each immutable variant
 
 
@dataclass(frozen=True)  # => one VARIANT of the Shape ADT -- only the field a circle needs
class Circle:  # => the Circle variant's body
    radius: float  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => a SECOND variant -- only the field a square needs
class Square:  # => the Square variant's body
    side: float  # => the single field this variant carries
 
 
Shape = Circle | Square  # => PEP 604 union syntax (3.10+): a Shape IS EITHER a Circle OR a Square
 
 
def area(shape: Shape) -> float:  # => accepts EITHER variant -- illegal combinations don't type-check
    if isinstance(shape, Circle):  # => narrows shape to Circle inside this branch
        return 3.14159 * shape.radius**2  # => the actual circle-area formula
    return shape.side**2  # => shape is narrowed to Square here (the only remaining case)
 
 
circle_shape: Shape = Circle(radius=2.0)  # => constructs the Circle variant
square_shape: Shape = Square(side=3.0)  # => constructs the Square variant
 
# => a sum type restricts callers to EXACTLY the variants the ADT defines
print(round(area(circle_shape), 2))  # => Output: 12.57
print(area(square_shape))  # => Output: 9.0
print(isinstance(circle_shape, Square))  # => Output: False -- a Circle can never ALSO be a Square

12.57
9.0
False

"""Example 45: pytest verification for A Shape ADT as a Union of Frozen Dataclasses."""
 
from example import Circle, Square, area
 
 
def test_each_variant_computes_its_own_area() -> None:
    assert round(area(Circle(radius=1.0)), 2) == 3.14
    assert area(Square(side=4.0)) == 16.0
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 46: match/case Dispatches Over the Shape ADT."""
 
from __future__ import annotations  # => enables the quoted forward references used below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds each immutable variant
 
 
@dataclass(frozen=True)  # => marks Circle immutable, matching the FP style
class Circle:  # => the Circle variant's body
    radius: float  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Square immutable, matching the FP style
class Square:  # => the Square variant's body
    side: float  # => the single field this variant carries
 
 
Shape = Circle | Square  # => the ADT itself: a Shape is EITHER variant
 
 
def describe(shape: Shape) -> str:  # => match/case, the structural-pattern-matching alternative to isinstance
    match shape:  # => picks the FIRST case whose pattern matches shape's shape
        case Circle(radius=r):  # => destructures a Circle, binding its radius to r
            return f"circle with radius {r}"  # => the Circle branch's result
        case Square(side=s):  # => destructures a Square, binding its side to s
            return f"square with side {s}"  # => the Square branch's result
        case _:  # => NOTE: match/case has NO compile-time exhaustiveness check -- a manual safety net
            raise ValueError("unreachable for a Circle | Square")  # => defends against a THIRD variant appearing later
 
 
# => match/case reads like the ADT's own shape, not a chain of isinstance checks
print(describe(Circle(radius=2.0)))  # => Output: circle with radius 2.0
print(describe(Square(side=5.0)))  # => Output: square with side 5.0

circle with radius 2.0
square with side 5.0

"""Example 46: pytest verification for match/case Dispatches Over the Shape ADT."""
 
from example import Circle, Square, describe
 
 
def test_each_branch_fires_for_its_own_variant() -> None:
    assert describe(Circle(radius=1.0)) == "circle with radius 1.0"
    assert describe(Square(side=2.0)) == "square with side 2.0"
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 47: case Clauses With if Guards."""
 
 
def classify(n: int) -> str:  # => match/case WITH guards -- a pattern PLUS a condition
    match n:  # => opens the match/case block over n
        case int(value) if value < 0:  # => guard: only matches negative ints
            return "negative"  # => the negative branch's result
        case 0:  # => an exact-value pattern, no guard needed
            return "zero"  # => the zero branch's result
        case int(value) if value % 2 == 0:  # => guard: only matches even positive ints
            return "positive even"  # => the positive-even branch's result
        case _:  # => catches everything else -- positive odd ints
            return "positive odd"  # => the positive-odd branch's result
 
 
# => guards let one case pattern cover many related conditions
print(classify(-5))  # => Output: negative
print(classify(0))  # => Output: zero
print(classify(4))  # => Output: positive even
print(classify(7))  # => Output: positive odd

negative
zero
positive even
positive odd

"""Example 47: pytest verification for case Clauses With if Guards."""
 
from example import classify
 
 
def test_guards_select_the_right_branch() -> None:
    assert classify(-1) == "negative"
    assert classify(0) == "zero"
    assert classify(2) == "positive even"
    assert classify(3) == "positive odd"
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 48: A Hand-Rolled Some/Nothing With map."""
 
from __future__ import annotations  # => enables the quoted 'Option[U]' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Option variants
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar make Some[T] a proper generic container
 
T = TypeVar("T")  # => the type of the value a Some wraps
U = TypeVar("U")  # => the type map's function returns
 
 
@dataclass(frozen=True)  # => marks Some immutable, matching the FP style
class Some(Generic[T]):  # => the "present" variant, carrying exactly one value
    value: T  # => the single field this variant carries
 
    def map(self, fn: Callable[[T], U]) -> "Option[U]":  # => applies fn INSIDE the wrapper
        return Some(fn(self.value))  # => stays wrapped -- Some in, Some out
 
 
@dataclass(frozen=True)  # => marks Nothing immutable too
class Nothing:  # => the "absent" variant, carrying nothing
    def map(self, fn: Callable[[T], U]) -> "Nothing":  # => NO-OP: still generic so it type-checks like Some.map
        return self  # => there is nothing to apply fn to -- Nothing stays Nothing
 
 
Option = Some[T] | Nothing  # => PEP 604 union: an Option[T] is either Some[T] or Nothing
 
 
present: Option[int] = Some(5)  # => an Option holding a real value
absent: Option[int] = Nothing()  # => an Option holding nothing at all
 
increment: Callable[[int], int] = lambda x: x + 1  # => explicit annotation pins T/U to int at both call sites
mapped_present = present.map(increment)  # => Some(5).map(+1) -- the function DOES run
mapped_absent = absent.map(increment)  # => Nothing.map(+1) -- the function is SKIPPED entirely
 
# => Option replaces None with a type the reader (and a type checker) can reason about
print(mapped_present)  # => Output: Some(value=6)
print(mapped_absent)  # => Output: Nothing()
print(mapped_absent == Nothing())  # => Output: True -- map never turns Nothing into Some

Some(value=6)
Nothing()
True

"""Example 48: pytest verification for A Hand-Rolled Some/Nothing With map."""
 
from typing import Callable
 
from example import Nothing, Some
 
 
def test_map_runs_on_some_and_is_skipped_on_nothing() -> None:
    increment: Callable[[int], int] = lambda x: x + 1
    assert Some(5).map(increment) == Some(6)
    assert Nothing().map(increment) == Nothing()
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 49: Chaining Option-Returning Lookups."""
 
from __future__ import annotations  # => enables the quoted 'Option[T]' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Option variants
from typing import Generic, TypeVar  # => Generic/TypeVar make Some[T] a proper generic container
 
T = TypeVar("T")  # => the type of the value a Some wraps
 
 
@dataclass(frozen=True)  # => marks Some immutable, matching the FP style
class Some(Generic[T]):  # => the "present" variant, carrying exactly one value
    value: T  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Nothing immutable too
class Nothing:  # => the "absent" variant
    pass  # => carries no data at all
 
 
Option = Some[T] | Nothing  # => the ADT itself: an Option is EITHER variant
 
 
def find(table: dict[str, T], key: str) -> "Option[T]":  # => an Option-returning lookup, no None
    return Some(table[key]) if key in table else Nothing()  # => success wraps, miss returns Nothing
 
 
def find_user_then_city(  # => opens the multi-line signature of the chained lookup
    users: dict[str, str], cities: dict[str, str], username: str  # => the two lookup tables plus the key to chase
) -> "Option[str]":  # => closes the multi-line signature above
    user_result = find(users, username)  # => first Option-returning lookup
    if isinstance(user_result, Nothing):  # => SHORT-CIRCUITS immediately on the first miss
        return Nothing()  # => never even attempts the second lookup
    return find(cities, user_result.value)  # => chains into a SECOND Option-returning lookup
 
 
users = {"ana": "jakarta"}  # => maps username -> city key
cities = {"jakarta": "Jakarta, Indonesia"}  # => maps city key -> display name
 
hit = find_user_then_city(users, cities, "ana")  # => both lookups succeed
miss = find_user_then_city(users, cities, "budi")  # => the FIRST lookup already misses
 
# => chaining Option lookups avoids nested None-checks entirely
print(hit)  # => Output: Some(value='Jakarta, Indonesia')
print(miss)  # => Output: Nothing()

Some(value='Jakarta, Indonesia')
Nothing()

"""Example 49: pytest verification for Chaining Option-Returning Lookups."""
 
from example import Nothing, Some, find_user_then_city
 
 
def test_chain_short_circuits_on_the_first_miss() -> None:
    users = {"ana": "jakarta"}
    cities = {"jakarta": "Jakarta, Indonesia"}
 
    assert find_user_then_city(users, cities, "ana") == Some("Jakarta, Indonesia")
    assert find_user_then_city(users, cities, "budi") == Nothing()
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 50: A Hand-Rolled Ok/Err Result Type."""
 
from __future__ import annotations  # => enables the quoted 'Result[float, str]' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Result variants
from typing import Generic, TypeVar  # => Generic/TypeVar make Ok[T] and Err[E] proper generic containers
 
T = TypeVar("T")  # => the type of the value an Ok wraps
E = TypeVar("E")  # => the type of the error an Err wraps
 
 
@dataclass(frozen=True)  # => marks Ok immutable, matching the FP style
class Ok(Generic[T]):  # => the "success" variant, carrying the computed value
    value: T  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Err immutable too
class Err(Generic[E]):  # => the "failure" variant, carrying the ERROR AS A VALUE, not an exception
    error: E  # => the single field this variant carries
 
 
Result = Ok[T] | Err[E]  # => PEP 604 union: a Result is either Ok[T] or Err[E]
 
 
def divide(a: int, b: int) -> "Result[float, str]":  # => the failure mode is IN the return type
    if b == 0:  # => a caller reading this signature already knows failure is possible
        return Err("division by zero")  # => the error travels as an ordinary VALUE
    return Ok(a / b)  # => success wraps the computed value
 
 
success = divide(10, 2)  # => Ok(5.0)
failure = divide(10, 0)  # => Err('division by zero') -- NO exception was raised
 
# => Result makes failure part of the TYPE, not a hidden exception path
print(success)  # => Output: Ok(value=5.0)
print(failure)  # => Output: Err(error='division by zero')
print(isinstance(failure, Err))  # => Output: True -- the caller can inspect this without a try/except

Ok(value=5.0)
Err(error='division by zero')
True

"""Example 50: pytest verification for A Hand-Rolled Ok/Err Result Type."""
 
from example import Err, Ok, divide
 
 
def test_result_carries_success_or_failure_as_a_value() -> None:
    assert divide(10, 2) == Ok(5.0)
    assert divide(10, 0) == Err("division by zero")
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 51: map and and_then on a Result."""
 
from __future__ import annotations  # => enables the quoted 'Result[U, object]' forward references below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Result variants
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar/Callable type the map and and_then methods
 
T = TypeVar("T")  # => the type of the value an Ok wraps
U = TypeVar("U")  # => the type map/and_then transform T into
E = TypeVar("E")  # => the type of the error an Err wraps
F = TypeVar("F")  # => the error type threaded through and_then's step function, not hardcoded to object
 
 
@dataclass(frozen=True)  # => marks Ok immutable, matching the FP style
class Ok(Generic[T]):  # => the success variant's body
    value: T  # => the single field this variant carries
 
    def map(self, fn: Callable[[T], U]) -> "Result[U, object]":  # => transforms the SUCCESS value only
        return Ok(fn(self.value))  # => transforms the value, stays wrapped as Ok
 
    def and_then(  # => chains into ANOTHER Result-returning step, without double-wrapping
        self, fn: Callable[[T], "Result[U, F]"]  # => F is inferred from fn's own error type, not widened to object
    ) -> "Result[U, F]":  # => the chained Result keeps fn's REAL error type
        return fn(self.value)  # => runs fn on the unwrapped value; fn itself returns a Result
 
 
@dataclass(frozen=True)  # => marks Err immutable too
class Err(Generic[E]):  # => the failure variant's body
    error: E  # => the single field this variant carries
 
    def map(self, fn: Callable[[T], U]) -> "Err[E]":  # => NO-OP: generic so a typed fn still passes the check
        return self  # => the error passes through UNCHANGED -- fn never runs
 
    def and_then(self, fn: Callable[[T], U]) -> "Err[E]":  # => and_then is ALSO a no-op on Err, generic likewise
        return self  # => short-circuits: fn never runs once the pipeline has already failed
 
 
Result = Ok[T] | Err[E]  # => the ADT itself: a Result is EITHER variant
 
 
def parse_positive(text: str) -> "Result[int, str]":  # => an and_then STEP: str -> Result[int, str]
    value = int(text)  # => may raise, but this example only feeds it valid ints
    return Ok(value) if value > 0 else Err(f"{value} is not positive")  # => the and_then step itself: may succeed or fail
 
 
times_ten: Callable[[int], int] = lambda n: n * 10  # => explicit annotation pins map's generic parameter to int
 
ok_chain = Ok("5").and_then(parse_positive).map(times_ten)  # => success end to end
err_chain = Ok("-5").and_then(parse_positive).map(times_ten)  # => fails at and_then
 
# => map transforms success; and_then chains into ANOTHER fallible step
print(ok_chain)  # => Output: Ok(value=50)
print(err_chain)  # => Output: Err(error='-5 is not positive')

Ok(value=50)
Err(error='-5 is not positive')

"""Example 51: pytest verification for map and and_then on a Result."""
 
from typing import Callable
 
from example import Err, Ok, parse_positive
 
 
def test_pipeline_stops_at_the_first_err() -> None:
    times_ten: Callable[[int], int] = lambda n: n * 10
    ok_chain = Ok("5").and_then(parse_positive).map(times_ten)
    err_chain = Ok("-5").and_then(parse_positive).map(times_ten)
 
    assert ok_chain == Ok(50)
    assert err_chain == Err("-5 is not positive")
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 52: A Validation Pipeline Threading Result."""
 
from __future__ import annotations  # => enables the quoted 'Result[int, str]' forward references below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Result variants
from typing import Generic, TypeVar  # => Generic/TypeVar make Ok[T] and Err[E] proper generic containers
 
T = TypeVar("T")  # => the type of the value an Ok wraps
E = TypeVar("E")  # => the type of the error an Err wraps
 
 
@dataclass(frozen=True)  # => marks Ok immutable, matching the FP style
class Ok(Generic[T]):  # => the success variant's body
    value: T  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Err immutable too
class Err(Generic[E]):  # => the failure variant's body
    error: E  # => the single field this variant carries
 
 
Result = Ok[T] | Err[E]  # => the ADT itself: a Result is EITHER variant
 
 
def validate_age(age: int) -> "Result[int, str]":  # => railway step 1: switches to the failure track on error
    if age < 0:  # => the ONLY failure condition this step checks
        return Err("age cannot be negative")  # => switches to the failure track
    return Ok(age)  # => the success track, value unchanged
 
 
def validate_name(name: str) -> "Result[str, str]":  # => railway step 2
    if not name:  # => the ONLY failure condition this step checks
        return Err("name cannot be empty")  # => switches to the failure track
    return Ok(name)  # => the success track, value unchanged
 
 
def validate_form(name: str, age: int) -> "Result[str, str]":  # => threads BOTH steps in sequence
    name_result = validate_name(name)  # => the FIRST switch point
    if isinstance(name_result, Err):  # => already on the failure track -- age is never checked
        return name_result  # => the original error rides through untouched
    age_result = validate_age(age)  # => the SECOND switch point, only reached if step 1 succeeded
    if isinstance(age_result, Err):  # => the SECOND switch point
        return age_result  # => propagates the second step's failure
    return Ok(f"{name_result.value}, age {age_result.value}")  # => both checks passed
 
 
valid = validate_form("Ana", 30)  # => both checks succeed
bad_name = validate_form("", 30)  # => fails at the FIRST check -- age is never even validated
bad_age = validate_form("Ana", -1)  # => passes name, fails at the SECOND check
 
# => this is the railway-oriented programming pattern: one track, two rails
print(valid)  # => Output: Ok(value='Ana, age 30')
print(bad_name)  # => Output: Err(error='name cannot be empty')
print(bad_age)  # => Output: Err(error='age cannot be negative')

Ok(value='Ana, age 30')
Err(error='name cannot be empty')
Err(error='age cannot be negative')

"""Example 52: pytest verification for A Validation Pipeline Threading Result."""
 
from example import Err, Ok, validate_form
 
 
def test_one_bad_field_short_circuits_the_rest() -> None:
    assert validate_form("Ana", 30) == Ok("Ana, age 30")
    assert validate_form("", 30) == Err("name cannot be empty")
    assert validate_form("Ana", -1) == Err("age cannot be negative")
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 53: The Functor Identity Law by Example."""
 
from __future__ import annotations  # => enables the quoted 'Box[U]' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds the immutable Box
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar/Callable type the minimal functor below
 
T = TypeVar("T")  # => the type of the value a Box wraps
U = TypeVar("U")  # => the type map's function returns
 
 
def identity(x: T) -> T:  # => the identity function: returns its argument UNCHANGED
    return x  # => identity's entire body: returns its argument, unchanged
 
 
@dataclass(frozen=True)  # => marks Box immutable, matching the FP style
class Box(Generic[T]):  # => a minimal functor: any container with a lawful map
    value: T  # => the single field this container carries
 
    def map(self, fn: Callable[[T], U]) -> "Box[U]":  # => applies fn inside, stays wrapped
        return Box(fn(self.value))  # => applies fn inside, re-wraps as Box
 
 
original = Box(42)  # => the container BEFORE the identity law is applied
mapped_with_identity = original.map(identity)  # => container.map(identity)
 
# => the functor identity law is a correctness check any lawful map must pass
print(mapped_with_identity == original)  # => Output: True -- the FUNCTOR IDENTITY LAW, verified
print(mapped_with_identity is original)  # => Output: False -- EQUAL, but a distinct object (still immutable)

True
False

"""Example 53: pytest verification for The Functor Identity Law by Example."""
 
from example import Box, identity
 
 
def test_mapping_identity_changes_nothing_but_the_wrapper() -> None:
    original = Box(7)
    assert original.map(identity) == original  # => the functor identity law
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 54: One fmap Working on list and Option."""
 
from __future__ import annotations  # => enables the quoted 'Option[T]' forward references below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Option variants
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar/Callable type the shared fmap below
 
T = TypeVar("T")  # => the type of the value a container wraps
U = TypeVar("U")  # => the type fn transforms T into
 
 
@dataclass(frozen=True)  # => marks Some immutable, matching the FP style
class Some(Generic[T]):  # => the present variant's body
    value: T  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Nothing immutable too
class Nothing:  # => the absent variant's body
    pass  # => carries no data at all
 
 
Option = Some[T] | Nothing  # => the ADT itself: an Option is EITHER variant
 
 
def fmap(fn: Callable[[T], U], container: "list[T] | Some[T] | Nothing") -> object:  # => ONE signature, THREE possible container shapes
    # => ONE function, dispatching on the container's actual shape -- the functor pattern made concrete
    if isinstance(container, list):  # => a list is mappable via a comprehension
        return [fn(item) for item in container]  # => applies fn to every element
    if isinstance(container, Some):  # => an Option is mappable via its own map rule
        return Some(fn(container.value))  # => applies fn to the single wrapped value
    return Nothing()  # => Nothing maps to Nothing, regardless of fn
 
 
double: Callable[[int], int] = lambda n: n * 2  # => explicit annotation lets fmap solve T=int from fn, not the lambda body
 
mapped_list = fmap(double, [1, 2, 3])  # => the list case
mapped_some = fmap(double, Some(5))  # => the Option case, present
mapped_nothing = fmap(double, Nothing())  # => the Option case, absent
 
print(mapped_list)  # => Output: [2, 4, 6]
print(mapped_some)  # => Output: Some(value=10)
print(mapped_nothing)  # => Output: Nothing()

[2, 4, 6]
Some(value=10)
Nothing()

"""Example 54: pytest verification for One fmap Working on list and Option."""
 
from typing import Callable
 
from example import Nothing, Some, fmap
 
 
def test_the_same_fmap_dispatches_on_container_shape() -> None:
    double: Callable[[int], int] = lambda n: n * 2
    assert fmap(double, [1, 2]) == [2, 4]
    assert fmap(double, Some(3)) == Some(6)
    assert fmap(double, Nothing()) == Nothing()
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 55: map2 Combines Two Options."""
 
from __future__ import annotations  # => enables the quoted 'Option[V]' forward reference below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Option variants
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar/Callable type the applicative map2 below
 
T = TypeVar("T")  # => the type of the first wrapped value
U = TypeVar("U")  # => the type of the second wrapped value
V = TypeVar("V")  # => the type fn returns after combining both
 
 
@dataclass(frozen=True)  # => marks Some immutable, matching the FP style
class Some(Generic[T]):  # => the present variant's body
    value: T  # => the single field this variant carries
 
 
@dataclass(frozen=True)  # => marks Nothing immutable too
class Nothing:  # => the absent variant's body
    pass  # => carries no data at all
 
 
Option = Some[T] | Nothing  # => the ADT itself: an Option is EITHER variant
 
 
def map2(  # => the applicative pattern: combine TWO wrapped values with a 2-arg function
    fn: Callable[[T, U], V], opt_a: "Option[T]", opt_b: "Option[U]"  # => the 2-arg combiner plus the two Options it combines
) -> "Option[V]":  # => closes the multi-line signature above
    if isinstance(opt_a, Nothing) or isinstance(opt_b, Nothing):  # => EITHER absent short-circuits
        return Nothing()  # => no partial combination -- both-or-nothing
    return Some(fn(opt_a.value, opt_b.value))  # => both present: unwrap both, apply fn, rewrap
 
 
def add(a: int, b: int) -> int:  # => the 2-argument function map2 combines two Options with
    return a + b  # => the actual addition add performs
 
 
both_present = map2(add, Some(2), Some(3))  # => both wrapped values ARE present
one_missing = map2(add, Some(2), Nothing())  # => the second value is ABSENT
 
# => applicative map2 generalizes map to functions of MORE than one argument
print(both_present)  # => Output: Some(value=5)
print(one_missing)  # => Output: Nothing()

Some(value=5)
Nothing()

"""Example 55: pytest verification for map2 Combines Two Options."""
 
from example import Nothing, Some, add, map2
 
 
def test_map2_combines_when_both_present_and_short_circuits_otherwise() -> None:
    assert map2(add, Some(2), Some(3)) == Some(5)
    assert map2(add, Some(2), Nothing()) == Nothing()
    assert map2(add, Nothing(), Some(3)) == Nothing()
 
 
# => Run: pytest -- Output: 1 passed

1 passed

"""Example 56: bind/flat_map Chaining Result Steps."""
 
from __future__ import annotations  # => enables the quoted 'Result[U, object]' forward references below
 
from dataclasses import dataclass  # => @dataclass(frozen=True) builds both Result variants
from typing import Callable, Generic, TypeVar  # => Generic/TypeVar/Callable type the bind chain below
 
T = TypeVar("T")  # => the type of the value an Ok wraps
U = TypeVar("U")  # => the type bind's step function returns
E = TypeVar("E")  # => the type of the error an Err wraps
F = TypeVar("F")  # => the error type threaded through bind's step function, not hardcoded to object
 
 
@dataclass(frozen=True)  # => marks Ok immutable, matching the FP style
class Ok(Generic[T]):  # => the success variant's body
    value: T  # => the single field this variant carries
 
    def bind(self, fn: Callable[[T], "Result[U, F]"]) -> "Result[U, F]":  # => F comes from fn's OWN error type
        # => bind/and_then/flat_map -- three names for the SAME monadic chaining operation
        return fn(self.value)  # => unwraps, runs fn (which itself returns a Result), no double-wrap
 
 
@dataclass(frozen=True)  # => marks Err immutable too
class Err(Generic[E]):  # => the failure variant's body
    error: E  # => the single field this variant carries
 
    def bind(self, fn: Callable[[T], U]) -> "Err[E]":  # => NO-OP, generic so a typed fn still passes the check
        return self  # => fn never runs once the chain has already failed
 
 
Result = Ok[T] | Err[E]  # => the ADT itself: a Result is EITHER variant
 
 
def half(n: int) -> "Result[float, str]":  # => a bind STEP: fails if n is odd
    if n % 2 != 0:  # => the ONLY failure condition this step checks
        return Err(f"{n} is odd")  # => switches to the failure track
    return Ok(n / 2)  # => the success track: n halved
 
 
def to_positive(x: float) -> "Result[float, str]":  # => a second bind STEP: fails if not positive
    return Ok(x) if x > 0 else Err(f"{x} is not positive")  # => the second bind step: may succeed or fail
 
 
chained_success = Ok(8).bind(half).bind(to_positive)  # => 8 -> 4.0 -> still positive
chained_failure = Ok(7).bind(half).bind(to_positive)  # => 7 is odd -- fails at the FIRST bind
 
print(chained_success)  # => Output: Ok(value=4.0)
print(chained_failure)  # => Output: Err(error='7 is odd')

Ok(value=4.0)
Err(error='7 is odd')

"""Example 56: pytest verification for bind/flat_map Chaining Result Steps."""
 
from example import Err, Ok, half, to_positive
 
 
def test_bind_chains_and_short_circuits_on_the_first_error() -> None:
    assert Ok(8).bind(half).bind(to_positive) == Ok(4.0)
    assert Ok(7).bind(half).bind(to_positive) == Err("7 is odd")
 
 
# => Run: pytest -- Output: 1 passed

1 passed