M05C03: Functors in Python – “Things You Can Map Over” (Option, Result, List)¶

Progression Note¶

By the end of Module 5, you will model every domain concept as immutable algebraic data types (products and tagged sums), eliminating whole classes of runtime errors through exhaustive pattern matching, mypy-checked totality, and pure serialization contracts.

Module	Focus	Key Outcomes
4	Safe Recursion & Error Handling	Stack-safe tree recursion, folds, Result/Option, streaming validation/retries
5	Advanced Type-Driven Design	ADTs, exhaustive pattern matching, total functions, refined types
6	Monadic Flows as Composable Pipelines	bind/and_then, Reader/State-like patterns, error-typed flows

Core question
How do you replace ad-hoc unboxing, duplicated if err/None boilerplate, and manual loops with lawful functor mapping over Option, Result, and List — guaranteeing type-safe, composable transformations in every FuncPipe pipeline stage?

We now take the immutable state ADTs from C02 and ask the question every pipeline eventually faces:

“Why do I have the same 6-line unboxing pattern copied into 50 functions, and why does a tiny refactor in one transformation break half the pipeline without mypy noticing?”

The naïve pattern everyone writes first:

# BEFORE – manual unboxing everywhere
def normalize_embedding(res: Result[Embedding, ErrInfo]) -> Result[float, ErrInfo]:
    if isinstance(res, Err):
        return res
    try:
        norm = normalize(res.value)
        return Ok(norm)
    except Exception as e:
        return Err(make_errinfo(...))

# And for batches...
def embed_batch(chunks: list[Chunk]) -> list[Result[Success, ErrInfo]]:
    out = []
    for c in chunks:
        r = embed_chunk(c)
        if isinstance(r, Err):
            out.append(r)
            continue
        out.append(Ok(postprocess(r.value)))
    return out

Duplicated boilerplate, mixed Optional/Result, manual loops, silent inconsistencies.

The production pattern: every container implements a curried lawful map that applies a pure function only when there is a value, preserving structure automatically.

# AFTER – uniform, composable, zero boilerplate
normalize_embedding = result_try_map(normalize, stage="norm")

postprocess = result_map(normalize_embedding) >> result_map(str)

embed_batch = iter_map(embed_chunk) >> iter_map(postprocess)
# or eager:
embed_batch = list_map(embed_chunk) >> list_map(postprocess)

Now every transformation is a single, composable, type-checked line — forever.

Audience: Engineers drowning in duplicated unboxing who want mathematically lawful, uniformly mappable containers.

Outcome 1. Every manual if Err/None or loop replaced with functor map. 2. All transformations proven to satisfy identity + composition laws. 3. Immutable, lazy-capable pipelines with zero boilerplate.

Tiny Non-Domain Example – Mapping Over Option[int]¶

opt: Option[int] = Some(value=5)
doubled = option_map(lambda x: x * 2)(opt)          # Some(value=10)
tripled = option_map(lambda x: x + x//2)(doubled)   # Some(value=15)

absent: Option[int] = NONE
option_map(lambda x: x * 100)(absent)               # NONE – function never called

Mapping applies only when present, short-circuits automatically, and composes cleanly.

Why Functors? (Three bullets every engineer should internalise)¶

Uniform transformation: One curried map works for Option, Result, List — no more duplicated unboxing.
Lawful composition: map(f) >> map(g) == map(f >> g) and map(id) == id proven for all instances — refactoring is safe.
Structure preservation + short-circuit: Errors/absence automatically propagated, no accidental double-processing.

We explicitly do not use None or Optional[T] as absence in domain code — absence is a first-class ADT (Option[T] = Some[T] | NONE) with its own lawful map.

1. Laws & Invariants (machine-checked)¶

Reminder – what does >> mean in this series?
Throughout Modules 1–5 we use f >> g as left-to-right function composition: first apply f, then apply g to the result. Formally,
f >> g := compose(f, g) := lambda x: g(f(x)).
In concrete Python you can always replace f >> g with compose(f, g) (or with the flow / pipe helpers introduced in earlier cores).

Law	Formal Statement	Enforcement
Identity	`map(id)(x) == x`	Hypothesis on all functors
Composition	`map(g)(map(f)(x)) == map(f >> g)(x)`	Hypothesis on all functors (`f >> g := compose(f, g)`)
Preservation	Mapping over empty/error yields empty/error	Property tests + short-circuit counters
Naturality	`to_optional ∘ map(f) == map(f) ∘ to_optional` (for Option)	Hypothesis bridge tests
Purity	`map` is referentially transparent	Reproducibility + no-side-effect tests
Bounded Work	O(1) for Option/Result, O(N) lazy for iter_map	Instrumented benchmarks

2. Decision Table – Which Functor When?¶

Container	May Be Absent	May Fail	Collection	Recommended Functor
Optional value	Yes	No	No	`Option`
Fallible op	No	Yes	No	`Result`
Sequence/Batch	No	No	Yes	`iter_map` (lazy) / `list_map` (eager tuple)

3. Public API (fp/functor.py – mypy --strict clean)¶

from __future__ import annotations

from dataclasses import dataclass
from typing import (Generic, TypeVar, Callable, Iterable, Iterator,
                    Tuple, Sequence, Literal, Optional, final, TypeAlias, cast)

from .core import Result, Ok, Err, ErrInfo, make_errinfo

__all__ = [
    "Option", "Some", "NoneVal", "NONE",
    "option_map", "from_optional", "to_optional",
    "result_map", "result_try_map", "result_map_err", "result_bimap",
    "iter_map", "list_map", "compose",
]

T = TypeVar("T")
U = TypeVar("U")
V = TypeVar("V")
E = TypeVar("E")
F = TypeVar("F")

# Left-to-right function composition.
# Throughout the cores we use the infix notation
#   f >> g  :=  compose(f, g)  :=  lambda x: g(f(x))
# i.e. “apply f, then g”. If you don’t have an infix helper
# in your codebase, call `compose(f, g)` directly (or use the
# `flow` / `pipe` helpers from earlier modules).
def compose(f: Callable[[T], U], g: Callable[[U], V]) -> Callable[[T], V]:
    return lambda x: g(f(x))

@dataclass(frozen=True, slots=True, kw_only=True)
class Some(Generic[T]):
    kind: Literal["some"] = "some"
    value: T

@final
@dataclass(frozen=True, slots=True)
class NoneVal:
    kind: Literal["none"] = "none"

NONE: NoneVal = NoneVal()  # singleton – do not construct NoneVal() directly
Option: TypeAlias = Some[T] | NoneVal

# Option functor (curried)
def option_map(f: Callable[[T], U]) -> Callable[[Option[T]], Option[U]]:
    def _inner(opt: Option[T]) -> Option[U]:
        return Some(value=f(opt.value)) if isinstance(opt, Some) else NONE
    return _inner

def from_optional(x: Optional[T]) -> Option[T]:
    return NONE if x is None else Some(value=x)

def to_optional(opt: Option[T]) -> Optional[T]:
    return opt.value if isinstance(opt, Some) else None

# Result functor (curried, generic in error type E)
def result_map(
    f: Callable[[T], U],
) -> Callable[[Result[T, E]], Result[U, E]]:
    def _inner(res: Result[T, E]) -> Result[U, E]:
        if isinstance(res, Ok):
            return Ok(value=f(res.value))
        # res is an Err here; value type stays T, we widen to U via cast
        return cast(Result[U, E], res)
    return _inner

def result_try_map(
    f: Callable[[T], U],
    *,
    stage: str = "",
    path: Tuple[int, ...] = (),
) -> Callable[[Result[T, ErrInfo]], Result[U, ErrInfo]]:
    def _inner(res: Result[T, ErrInfo]) -> Result[U, ErrInfo]:
        if isinstance(res, Err):
            return res
        try:
            return Ok(value=f(res.value))
        except Exception as exc:
            return Err(error=make_errinfo(
                code="EXC",
                msg=str(exc),
                stage=stage,
                path=path,
                exc=exc,
                meta={"exc_type": type(exc).__name__},
            ))
    return _inner

def result_map_err(
    f: Callable[[E], F],
) -> Callable[[Result[T, E]], Result[T, F]]:
    def _inner(res: Result[T, E]) -> Result[T, F]:
        if isinstance(res, Err):
            return Err(error=f(res.error))
        return cast(Result[T, F], res)
    return _inner

def result_bimap(
    f: Callable[[T], U],
    g: Callable[[E], F],
) -> Callable[[Result[T, E]], Result[U, F]]:
    def _inner(res: Result[T, E]) -> Result[U, F]:
        if isinstance(res, Ok):
            return Ok(value=f(res.value))
        # res is an Err here; we transform its error payload
        err = cast(Err[E], res)
        return Err(error=g(err.error))
    return _inner

# Sequence functors (curried)
def iter_map(f: Callable[[T], U]) -> Callable[[Iterable[T]], Iterator[U]]:
    def _inner(xs: Iterable[T]) -> Iterator[U]:
        for x in xs:
            yield f(x)
    return _inner

# Returns immutable tuple for value semantics
def list_map(f: Callable[[T], U]) -> Callable[[Sequence[T]], Tuple[U, ...]]:
    def _inner(xs: Sequence[T]) -> Tuple[U, ...]:
        return tuple(iter_map(f)(xs))
    return _inner

4. Reference Implementations (continued)¶

4.1 Before vs After – Result Mapping¶

# BEFORE – 8 lines of boilerplate
def normalize_embedding(res: Result[Embedding, ErrInfo]) -> Result[float, ErrInfo]:
    if isinstance(res, Err):
        return res
    try:
        norm = normalize(res.value)
        return Ok(value=norm)
    except Exception as e:
        return Err(error=make_errinfo(...))

# AFTER – one lawful line
normalize_embedding = result_try_map(normalize, stage="norm")

4.2 Before vs After – Batch Processing¶

# BEFORE – manual loop, error handling mixed in
results = []
for chunk in chunks:
    r = embed_chunk(chunk)
    if isinstance(r, Err):
        results.append(r)
        continue
    results.append(Ok(value=postprocess(r.value)))

# AFTER – pure, lazy, composable
pipeline = iter_map(embed_chunk) >> iter_map(postprocess)
results = tuple(pipeline(chunks))

4.3 RAG Integration Example¶

# Full embedding pipeline – zero unboxing
embed_and_postprocess = (
    result_try_map(extract_text, stage="extract") >>
    result_try_map(tokenize, stage="tokenize") >>
    result_try_map(model.encode, stage="encode") >>
    result_try_map(normalize, stage="norm")
)

batch_pipeline = iter_map(embed_and_postprocess)
embedded_chunks: Tuple[Result[NormalizedEmbedding, ErrInfo], ...] = tuple(batch_pipeline(raw_chunks))

5. Property-Based Proofs (tests/test_functor_laws.py)¶

from dataclasses import dataclass
from hypothesis import given, strategies as st
from funcpipe_rag.fp.functor import *
from funcpipe_rag.fp.core import Ok, Err

# Simple dummy error for law tests (decoupled from full ErrInfo)
@dataclass(frozen=True)
class DummyErr:
    msg: str

@given(x=st.integers())
def test_option_identity(x):
    opt = Some(value=x)
    assert option_map(lambda v: v)(opt) == opt
    assert option_map(lambda v: v)(NONE) is NONE

@given(x=st.integers())
def test_option_composition(x):
    f = lambda v: v + 1
    g = lambda v: v * 2
    opt = Some(value=x)
    assert (option_map(g)(option_map(f)(opt)) ==
            option_map(compose(f, g))(opt))

@given(opt=st.one_of(st.builds(Some, value=st.integers()), st.just(NONE)))
def test_option_naturality(opt):
    f = lambda v: v + 5
    assert to_optional(option_map(f)(opt)) == (None if opt is NONE else f(opt.value))

@given(res=st.one_of(st.builds(Ok, value=st.integers()),
                    st.builds(Err, error=st.builds(DummyErr, msg=st.text()))))
def test_result_functor_laws(res):
    assert result_map(lambda x: x)(res) == res
    f = lambda x: x + 1
    g = lambda x: x * 2
    assert (result_map(g)(result_map(f)(res)) ==
            result_map(compose(f, g))(res))

@given(xs=st.lists(st.integers()))
def test_list_functor_laws(xs):
    assert list_map(lambda x: x)(xs) == tuple(xs)
    f = lambda x: x + 1
    g = lambda x: x * 2
    assert (list_map(g)(list_map(f)(xs)) ==
            list_map(compose(f, g))(xs))

@given(xs=st.lists(st.integers()))
def test_iter_functor_laws(xs):
    f = lambda x: x + 1
    g = lambda x: x * 2
    assert (tuple(iter_map(g)(iter_map(f)(xs))) ==
            tuple(iter_map(compose(f, g))(xs)))
    assert tuple(iter_map(lambda x: x)(xs)) == tuple(xs)

6. Big-O & Allocation Guarantees¶

Functor	Time	Heap (eager)	Heap (lazy)	Notes
option_map	O(1)	O(1)	–	Constant
result_map	O(1)	O(1)	–	Constant
iter_map	O(N)	O(1)	O(1)	Fully lazy
list_map	O(N)	O(N)	–	Immutable tuple output

7. Anti-Patterns & Immediate Fixes¶

Anti-Pattern	Symptom	Fix
Manual if Err/None	Duplicated boilerplate	`map` / `try_map` (curried)
List comprehensions with if	Mixed logic, hard to compose	`iter_map` / `list_map`
Mutable accumulation	Side effects	Pure `map` over immutable containers
Eager processing of streams	OOM	`iter_map` for laziness
Ignoring functor laws	Refactor breaks silently	Hypothesis law tests in CI

8. Pre-Core Quiz¶

Functor = container + lawful what? → curried map
Identity law → map(id)(x) == x
Composition law → map(g)(map(f)(x)) == map(f >> g)(x)
For absent/error values → map short-circuits automatically
For large sequences → use iter_map for laziness

9. Post-Core Exercise¶

Implement your own functor (e.g. Tree) → prove identity + composition.
Refactor one pipeline loop to iter_map(f) >> iter_map(g).
Replace all manual if res.is_err(): return res with result_map / result_try_map.
Add Hypothesis law tests for any custom container you map over.

Next: M05C04 – Applicative Validation (Independent Checks, Accumulate All Errors).

You now transform data through any container with a single, lawful, composable curried map — no unboxing, no loops, no boilerplate, short-circuit guaranteed. The rest of Module 5 builds Applicatives (parallel validation), Monads (sequencing), and Monoids (aggregation) on top of these functors.