"Truth is more easily concluded from error than from confusion." — Francis Bacon

Most software is built on a foundation of hope—the hope that the "Happy Path" will prevail. But in distributed systems, the Happy Path is a statistical anomaly. Chaos is the default.

combinators.py is a framework for building asynchronous systems that don't just "work," but work predictably. It provides a set of mathematical primitives (combinators) to transform fragile code into a resilient, type-safe computation graph.

❌ Standard Python: 25 lines of chaos

async def fetch_with_retry(url: str) -> Data:
    last_error = None
    for attempt in range(3):
        try:
            async with asyncio.timeout(5.0):
                result = await http.get(url)
                if result.status == 429:
                    await asyncio.sleep(2 ** attempt)
                    continue
                return result.json()
        except asyncio.TimeoutError:
            last_error = TimeoutError()
        except Exception as e:
            last_error = e
            await asyncio.sleep(2 ** attempt)
    
    try:
        return await cache.get(url)  # Fallback
    except:
        return DEFAULT_VALUE

✅ Combinators: 9 lines of intent

from combinators import flow, lift as L, fallback_chain
from combinators.control import RetryPolicy

fetch = (
    flow(L.call_catching(http.get, on_error=map_err, url=url))
    .retry(policy=RetryPolicy.exponential_jitter(times=3))
    .timeout(seconds=5.0)
    .compile()
)

result = await fallback_chain(fetch, cache_op, L.up.pure(DEFAULT))

The difference? One is instructions. The other is topology.

When we build reliable systems, we are often forced to write "glue code": retry loops, timeout wrappers, and nested try/except blocks. This code is imperative, repetitive, and often hides the actual business logic.

combinators.py turns these reliability patterns into first-class values.

• Retry is not a loop; it is a function.
• Timeout is not an exception; it is a type transformation.
• Fallback is not a nested branch; it is an algebraic choice.

By treating these patterns as values, we can compose them like LEGO bricks, building complex survival strategies from simple, honest building blocks.

Consider a multi-provider LLM pipeline. We want to race two providers, retry on rate limits, enforce an SLA, and guarantee content safety. In standard Python, this is a 50-line mess of concurrency and error handling. In combinators.py, it is a blueprint:

from combinators import flow, lift as L, race_ok
from combinators.control import RetryPolicy

# 1. Lift external APIs into an honest context
fetch_openai = L.call_catching(
    openai.chat.completions.create,
    on_error=map_openai_error,
    model="gpt-4",
    messages=[{"role": "user", "content": prompt}],
)

fetch_anthropic = L.call_catching(
    anthropic.messages.create,
    on_error=map_anthropic_error,
    model="claude-3",
    messages=[{"role": "user", "content": prompt}],
)

# 2. Compose the resilience strategy (The Blueprint)
pipeline = (
    flow(race_ok(fetch_openai, fetch_anthropic))  # Fastest success wins
    .retry(
        policy=RetryPolicy.exponential_jitter(times=3),
        retry_on=lambda e: e.kind == "rate_limit",  # Targeted retry
    )
    .timeout(seconds=10.0)                        # Enforce SLA
    .ensure(                                      # Domain invariant
        predicate=lambda r: r.content.safety_score > 0.9,
        error=lambda r: ContentPolicyError(score=r.content.safety_score),
    )
    .compile()
)

# 3. Execute with total type-safety
result = await pipeline  # Result[Message, APIError | TimeoutError | ContentPolicyError]

match result:
    case Ok(msg): 
        return Response(content=msg.text)
    case Error(ContentPolicyError(score)):
        return Response(error="unsafe_content", score=score, status=400)
    case Error(err):
        return Response(error="service_unavailable", details=str(err), status=503)

Standard programming assumes a "Local World." In the local world, memory is cheap, latency is zero, and functions always return.

Distributed systems exist in the "Global Chaos." In this world:

1. Latency is a feature, not a bug. If you don't model time, time will destroy your system.
2. Partial success is a failure in disguise. If two out of three nodes succeed, is your system in a valid state?
3. Observability is not optional. If you can't see why a computation failed, you can't fix it.

combinators.py is designed for the Global Chaos. It treats these problems not as annoyances to be patched, but as fundamental properties of the computation.

In standard Python, exceptions are invisible. You cannot see that fetch_user might raise a TimeoutError without reading its source code.

In combinators.py, reliability is driven by the type system. When you add a .timeout(2.0) to your flow, the type signature of your pipeline changes. It evolves from Interp[T, E] to Interp[T, E | TimeoutError].

This means the compiler becomes your partner in reliability. It prevents you from forgetting to handle a timeout. It forces you to acknowledge every failure mode you've introduced. This is what we call Type-Driven Reliability.

Standard Python functions lie. They claim to return a value but often throw an exception instead. combinators.py uses Result[T, E] to force functions to be honest about their failure modes. If a function can fail, it must say so in its type signature.

A Result represents the past. An Interp (Blueprint) represents the future. By making computations lazy, we gain the power to modify them before they run. We can add retries to a future that hasn't happened yet.

Computations are values. You can add them together (fallback), multiply them (parallel), or sequence them (then). These aren't metaphors; they are mathematical operations that preserve type safety and predictability.

We don't "validate" data after it enters our system; we parse it at the border. By the time your business logic receives a value, it is guaranteed to be valid by its very type. We make illegal states unrepresentable.

combinators.py is built on three levels of power, each tradeable for simplicity:

Distributed systems thrive on redundancy. Here is a production-grade read path: Primary DB raced against a Replica, falling back to a Cache, and finally to a Hardcoded Default.

from combinators import flow, lift as L, fallback_chain, race_ok
from combinators.control import RetryPolicy

# Define our tiers
primary = L.call_catching(db.get, on_error=map_db_error, key="config")
replica = L.call_catching(replica.get, on_error=map_db_error, key="config")
cache   = L.call_catching(redis.get, on_error=map_cache_error, key="config")

# Build the hierarchy
db_tier = (
    flow(race_ok(primary, replica)) # Low-latency dual-query
    .retry(times=2)                 # Transient retry
    .timeout(seconds=2.0)           # SLA enforcement
    .compile()
)

# The final read path: absolute resilience
read_path = fallback_chain(
    db_tier, 
    cache, 
    L.pure(DEFAULT_CONFIG) # The ultimate safety net
)

result = await read_path # Result[Config, Never]

combinators.py requires Python 3.13+ and the kungfu library.

uv add git+https://github.com/prostomarkeloff/combinators.py.git