Functors

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Algebraic Data Modelling Validation"]
  page["Functors"]
  capstone["Capstone evidence"]

  family --> program --> section --> page
  page -.applies in.-> capstone

flowchart LR
  orient["Orient on the page map"] --> read["Read the main claim and examples"]
  read --> inspect["Inspect the related code, proof, or capstone surface"]
  inspect --> verify["Run or review the verification path"]
  verify --> apply["Apply the idea back to the module and capstone"]

Functors should feel like a relief from repetitive container plumbing. The point is not more abstraction for its own sake. The point is a safe way to transform values inside Option, Result, and collections without rewriting the same structural boilerplate every time.

Start With the Unboxing Smell

The pain point here is easy to recognize: the actual transformation is tiny, but the code around it keeps checking Err, Some, empty cases, and manual loops.

• If the same if Err or if Some structure appears in many places, the transformation logic is being drowned by container handling.
• If a minor refactor changes structure-handling inconsistently across functions, the codebase no longer has one reliable transformation pattern.
• If you cannot describe what stays the same and what changes under map, the abstraction has not yet become useful.

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?

This lesson introduces functors as the standard answer to “transform the value, keep the container contract”:

• transform values without rewriting error, absence, or collection structure
• keep the container behavior stable while the inner function changes
• use the functor laws as a way to trust refactors, not as decorative mathematics

The motivating pipeline examples matter because they show the same smell in several shapes: optional values, fallible values, and batches all repeating structural handling.

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

This is the structural repetition to eliminate.

The production pattern makes the transformation itself the center of attention and lets the container preserve its own 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 the value transformation becomes the readable part again, and later composition has fewer places to drift apart.

Use this when duplicated unboxing is getting in the way and you want one consistent way to transform optional, fallible, and batched values.

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 (capstone/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

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

9. Post-Core Exercise

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

Continue with: Applicative Validation

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.