Skip to content

M06C01: and_then / bind Patterns for Result/Option Pipelines

Progression Note

Module 6 shifts from pure data modelling to effect-aware composition.
We now treat failure and absence as first-class effects that propagate automatically through pipelines — eliminating nested conditionals forever.

Module Focus Key Outcomes
5 Algebraic Data Modelling ADTs, exhaustive pattern matching, total functions, refined types
6 Monadic Flows as Composable Pipelines bind/and_then, Reader/State-like patterns, error-typed flows
7 Effect Boundaries & Resource Safety Dependency injection, boundaries, testing, evolution

Core question
How do you replace nested if err/None checks and manual error propagation with monadic and_then (a.k.a. bind or flat_map) — chaining dependent fallible/optional steps while automatically short-circuiting on the first failure or absence?

We take the pure ADTs from Module 5 and ask the question every pipeline eventually faces:
“Why do I have 47 copies of the same 8-line if res.is_err(): return res boilerplate, and why does adding a new step silently break half the pipeline?”

The naïve pattern everyone writes first: 

# BEFORE – manual propagation everywhere
def embed_chunk(c: Chunk) -> Result[EmbeddedChunk, ErrInfo]:
    tokenized = tokenize(c.text.content)
    if tokenized.is_err():
        return tokenized
    encoded = model.encode(tokenized.value)
    if encoded.is_err():
        return encoded
    return Ok(replace(c, embedding=Embedding(encoded.value, model.name)))

Duplicated checks, easy to forget a propagation, no compiler help.

The production pattern: every fallible or optional step returns a container, and and_then chains them — automatically short-circuiting on the first failure/absence.

Repo note: examples in this module use the Module-05 "compositional domain model" types: ChunkText.content / Embedding via from funcpipe_rag.rag.domain import Chunk, Embedding. The main RAG pipeline value types live separately in funcpipe_rag.core.rag_types (those chunks use text: str).

# AFTER – one lawful, composable pipeline (instance methods = canonical style)
def embed_chunk(c: Chunk) -> Result[EmbeddedChunk, ErrInfo]:
    return (
        Ok(c.text.content)
        .and_then(tokenize)
        .and_then(model.encode)
        .and_then(lambda vec: Ok(replace(c, embedding=Embedding(vec, model.name))))
    )

Note: the original c and model are captured via closure here (perfectly valid). We will replace this exact pattern with a Reader monad in M06C04 so dependencies are explicit and testable.

Adding a new step or changing error handling is a single-line change — forever.

Audience: Engineers tired of manual error propagation who want mathematically lawful, automatically short-circuiting pipelines.

Outcome 1. Every nested if err/None replaced with and_then. 2. All pipelines proven to satisfy monad laws (left/right identity, associativity). 3. Linear, readable, refactor-safe flows.

Tiny Non-Domain Example – Safe Division Chain

def safe_div(a: float, b: float) -> Result[float, str]:
    return Err("div by zero") if b == 0 else Ok(a / b)

chain = (
    Ok(12.0)
    .and_then(lambda x: safe_div(x, 4.0))
    .and_then(lambda x: safe_div(x, 3.0))
    .and_then(lambda x: safe_div(x, 0.0))   # short-circuits here
)
# → Err("div by zero")

The failing step short-circuits automatically — no manual checks.

Why and_then? (Three bullets every engineer should internalise)

  • Automatic short-circuit: First Err/NoneVal immediately propagates — no forgotten error paths.
  • Lawful composition: m.and_then(f).and_then(g) == m.and_then(lambda x: f(x).and_then(g)) — refactoring is safe.
  • Linear happy path: Success case reads top-to-bottom with no nesting — the code finally matches the mental model.

We explicitly do not use Optional[T] as Option[T] — absence is a first-class ADT (Some[T] | NoneVal) with its own lawful and_then.

1. Laws & Invariants (machine-checked)

Law Formal Statement Enforcement
Left Identity Ok(x).and_then(f) == f(x) Hypothesis
Right Identity m.and_then(Ok) == m Hypothesis
Associativity m.and_then(f).and_then(g) == m.and_then(lambda x: f(x).and_then(g)) Hypothesis
Short-circuit Err(e).and_then(f) == Err(e) / NoneVal().and_then(f) == NoneVal() Hypothesis + runtime

2. Decision Table – When to Use Which Container

Scenario May fail? May be absent? Accumulate errors? Use
API call / parsing Yes No No Result + and_then
Lookup / config key No Yes No Option + and_then
Multi-field validation Yes No Yes Validation + v_ap (see M06C03)

3. Public API – Instance methods are the canonical, idiomatic API

Type-system note: Python’s type system cannot express that the error branch of Result[T, E] is independent of the success type parameter T. In the actual codebase (src/funcpipe_rag/result/types.py) we keep mypy --strict green without type: ignore by re-wrapping the error value (e.g. returning Err(self.error)) in methods like map / and_then when the success type changes.

# src/funcpipe_rag/result/types.py – end-of-Module-06 (excerpted)

from __future__ import annotations
from dataclasses import dataclass
from typing import Generic, Callable, TypeVar, Any

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

# ==================== Result ====================

@dataclass(frozen=True)
class Ok(Generic[T, E]):
    value: T

    def and_then(self, f: Callable[[T], "Result[U, E]"]) -> "Result[U, E]":
        return f(self.value)

    def map(self, f: Callable[[T], U]) -> "Result[U, E]":
        return Ok(f(self.value))

    def or_else(self, _: Callable[[E], T]) -> T:
        return self.value

    def tap(self, side: Callable[[T], None]) -> "Ok[T, E]":
        side(self.value)
        return self

@dataclass(frozen=True)
class Err(Generic[T, E]):
    error: E

    def and_then(self, _: Callable[[T], "Result[U, E]"]) -> "Result[U, E]":
        return Err(self.error)

    def map(self, _: Callable[[T], U]) -> "Err[U, E]":
        return Err(self.error)

    def or_else(self, op: Callable[[E], T]) -> T:
        return op(self.error)

    def tap(self, _: Callable[[T], None]) -> "Err[T, E]":
        return self

Result = Ok[T, E] | Err[T, E]

# ==================== Option ====================

@dataclass(frozen=True)
class Some(Generic[T]):
    value: T

    def and_then(self, f: Callable[[T], "Option[U]"]) -> "Option[U]":
        return f(self.value)

    def map(self, f: Callable[[T], U]) -> "Option[U]":
        return Some(f(self.value))

    def or_else(self, op: Callable[[], T]) -> T:
        return self.value

    def tap(self, side: Callable[[T], None]) -> "Some[T]":
        side(self.value)
        return self

@dataclass(frozen=True)
class NoneVal:
    def and_then(self, _: Callable[[Any], "Option[U]"]) -> "Option[U]":
        return self

    def map(self, _: Callable[[Any], U]) -> "Option[U]":
        return self

    def or_else(self, op: Callable[[], T]) -> T:
        return op()

    def tap(self, _: Callable[[Any], None]) -> "NoneVal":
        return self

Option = Some[T] | NoneVal

Free-function wrappers exist only as one-liners for stylistic preference or Kleisli composition: 

def result_and_then(r: Result[T, E], f: Callable[[T], Result[U, E]]) -> Result[U, E]:
    return r.and_then(f)

Policy: In this series we use the instance method syntax everywhere (value.and_then(f)). It is the most readable and idiomatic. Free functions are purely optional.

4. Reference Implementations

4.1 RAG Integration – Embedding Pipeline (real-world closure usage)

def embed_chunk(c: Chunk) -> Result[EmbeddedChunk, ErrInfo]:
    return (
        Ok(c.text.content)
        .and_then(tokenize)
        .and_then(model.encode)
        .and_then(lambda vec: Ok(replace(c, embedding=Embedding(vec, model.name))))
    )

4.2 Config Parsing – Pure Result chain (absence = error here)

parse_config = lambda raw: (
    safe_json_load(raw)                                      # -> Result[dict, ErrInfo]
    .and_then(ensure(is_dict, "NOT_DICT"))
    .and_then(require_key("port", "MISSING_PORT"))
    .and_then(get("port"))
    .and_then(ensure(is_int, "BAD_PORT"))
    .and_then(ensure(in_range(1, 65535), "PORT_OOR"))
    .and_then(build_config)
)

All steps return Result[_, ErrInfo]. Missing key = error, not mere absence → stays cleanly in one monad.

5. Property-Based Proofs (tests/test_monad_laws.py)

from hypothesis import given, strategies as st
from funcpipe_rag.result.types import Ok, Err, Some, NoneVal, Result, Option

def st_result(t: st.SearchStrategy, e: st.SearchStrategy):
    return st.one_of(st.builds(Ok, t), st.builds(Err, e))

def st_option(t: st.SearchStrategy):
    return st.one_of(st.builds(Some, t), st.just(NoneVal()))

@given(x=st.integers())
def test_result_left_identity(x):
    f = lambda v: Ok(v * 2)
    assert Ok(x).and_then(f) == f(x)

@given(m=st_result(st.integers(), st.text()))
def test_result_right_identity(m):
    assert m.and_then(Ok) == m

@given(m=st_result(st.integers(), st.text()))
def test_result_associativity(m):
    f = lambda a: Ok(a + 1)
    g = lambda b: Ok(b * 2)
    assert m.and_then(f).and_then(g) == m.and_then(lambda a: f(a).and_then(g))

@given(e=st.text())
def test_result_short_circuit(e):
    assert Err(e).and_then(lambda _: Ok(999)) == Err(e)

@given(m=st_result(st.integers(), st.text()))
def test_result_tap_transparent(m):
    seen = []
    out = m.tap(lambda v: seen.append(v))
    assert out == m
    assert seen == ([m.value] if isinstance(m, Ok) else [])

@given(x=st.integers())
def test_option_left_identity(x):
    f = lambda v: Some(v * 2)
    assert Some(x).and_then(f) == f(x)

@given(o=st_option(st.integers()))
def test_option_right_identity(o):
    assert o.and_then(Some) == o

@given(o=st_option(st.integers()))
def test_option_associativity(o):
    f = lambda a: Some(a + 1) if a % 2 == 0 else NoneVal()
    g = lambda b: Some(b * 2)
    assert o.and_then(f).and_then(g) == o.and_then(lambda a: f(a).and_then(g))

@given(o=st_option(st.integers()))
def test_option_tap_transparent(o):
    seen = []
    out = o.tap(lambda v: seen.append(v))
    assert out == o
    assert seen == ([o.value] if isinstance(o, Some) else [])

6. Big-O & Allocation Guarantees

Operation Time Heap Notes
and_then / map / tap O(1) O(1) Combinators themselves; total cost dominated by user function f (may allocate new container)

7. Anti-Patterns & Immediate Fixes

Anti-Pattern Symptom Fix
Nested if/try/except Verbose, easy to miss propagation Chain with and_then
Manual flattening Nested Result/Option and_then auto-flattens
Side effects in the function to and_then Impure pipelines, non-reproducible Keep pure; use tap only for observation/logging
Ignoring laws Refactor silently breaks Hypothesis law tests in CI

8. Pre-Core Quiz

  1. and_then is for…?
    → Chaining dependent fallible/optional steps

  2. Short-circuits on…?
    → First Err/NoneVal

  3. Left identity law?
    → Ok(x).and_then(f) == f(x)

  4. Associativity enables…?
    → Refactor-safe composition (parentheses don’t matter)

  5. Never use … as a monad?
    → Optional[T] (builtin None)

9. Post-Core Exercise

  1. Take one real try/except + manual propagation chain in your codebase and rewrite it with and_then.
  2. Add a new fallible step to that chain and confirm only one line changes.
  3. Implement and_then (and the four laws) for a custom two-variant ADT of your own.
  4. Find and eliminate one instance of nested Result[Result[...]] or Option[Option[...]] using and_then.

Next: M06C02 – Law-Guided Design – Identity & Associativity Checks with Hypothesis.

You now chain any number of fallible or optional steps with a single, lawful and_then — short-circuiting automatically, no boilerplate, refactor-safe forever. The rest of Module 6 builds Reader/State/Writer patterns and configurable pipelines on top of these monads.