Result Pattern (zuidvolt/result-pattern)

Result Pattern

https://github.com/zuidvolt/result-pattern
Admin
Result Pattern is a Python library implementing the 'Errors as Values' pattern with Result and...

Tokens:10,925
Snippets:81
Trust Score:5.3
Update:2 months ago
Show doc for...
Context Summary (auto-generated)
Raw
# Result Pattern

Result Pattern is a Python library implementing the "Errors as Values" pattern for Python 3.14+. It provides Sum Types (`Result` with `Ok`/`Err` variants) and Product Types (`Outcome`) that make error handling explicit, type-safe, and composable. The library bridges functional programming paradigms with Python's exception-heavy ecosystem through lifting tools like `@catch` decorators, context managers, and monadic do-notation that automatically handles short-circuiting logic.

The core design philosophy centers on explicit error handling where failure paths become first-class API citizens. Unsafe operations that could panic are isolated in a `.unsafe` namespace, encouraging safe functional patterns. The library provides full support for modern Python features including pattern matching, async/await, and advanced type narrowing with tools like Basedpyright and Mypy. Combinators enable complex workflows like error accumulation, concurrent task orchestration, and fault-tolerant batch processing.

## Result Type - Ok and Err Variants

The `Result` type is a Sum Type representing either success (`Ok`) or failure (`Err`). It's the foundation for explicit error handling that makes failure modes visible in function signatures.

```python
from result import Ok, Err, Result, is_ok, is_err

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

# Type-safe narrowing with is_ok/is_err
res = divide(10, 2)
if is_ok(res):
    print(f"Success: {res.ok()}")  # Output: Success: 5.0

res2 = divide(10, 0)
if is_err(res2):
    print(f"Error: {res2.err()}")  # Output: Error: Cannot divide by zero

# Pattern matching (Python 3.10+)
match divide(10, 0):
    case Ok(val):
        print(f"Value: {val}")
    case Err(msg):
        print(f"Error: {msg}")  # Output: Error: Cannot divide by zero
```

## Functional Chaining with map and and_then

Transform values inside Results without unwrapping them. `map` applies a function to Ok values, while `and_then` chains operations that return Results (flatMap/bind).

```python
from result import Ok, Err, Result

def parse_int(s: str) -> Result[int, str]:
    try:
        return Ok(int(s))
    except ValueError:
        return Err(f"Invalid integer: {s}")

def validate_positive(n: int) -> Result[int, str]:
    return Ok(n) if n > 0 else Err("Must be positive")

# map: transform the success value
result = Ok(10).map(lambda x: x * 2)  # Ok(20)

# map_err: transform the error value
result = Err(404).map_err(lambda code: f"Code: {code}")  # Err('Code: 404')

# and_then: chain fallible operations
result = (
    parse_int("42")
    .and_then(validate_positive)
    .map(lambda x: x * 2)
)
print(result)  # Ok(84)

# Short-circuits on first error
result = (
    parse_int("-5")
    .and_then(validate_positive)
    .map(lambda x: x * 2)
)
print(result)  # Err('Must be positive')
```

## @catch Decorator and Context Manager

The `@catch` decorator lifts exception-throwing functions into Result-returning functions. It can also be used as a context manager for blocks of code.

```python
from result import catch, catch_call
from enum import StrEnum
import json

class ErrorCode(StrEnum):
    INVALID = "invalid_input"
    PARSE_ERROR = "parse_error"

# Basic catch - wraps exceptions in Err
@catch(ValueError)
def parse_int(s: str) -> int:
    return int(s)

print(parse_int("10"))    # Ok(10)
print(parse_int("abc"))   # Err(ValueError("invalid literal..."))

# Map exceptions to domain errors with map_to
@catch(ValueError, map_to=ErrorCode.INVALID)
def strict_parse(s: str) -> int:
    return int(s)

print(strict_parse("abc"))  # Err(<ErrorCode.INVALID: 'invalid_input'>)

# Multiple exception mapping with dictionary
@catch({ValueError: ErrorCode.INVALID, json.JSONDecodeError: ErrorCode.PARSE_ERROR})
def process_json(data: str) -> dict:
    return json.loads(data)

# Inline execution with catch_call
result = catch_call(json.JSONDecodeError, json.loads, '{"key": "value"}')
print(result)  # Ok({'key': 'value'})

# Context manager usage
with catch(json.JSONDecodeError) as ctx:
    data = json.loads('{"name": "Alice"}')
    ctx.set(data["name"])

print(ctx.result)  # Ok('Alice')
```

## do_notation - Monadic Do-Notation

The `@do_notation` decorator enables imperative-style code that automatically handles short-circuiting. Use `yield` to unwrap Results, and the function returns early on first Err.

```python
from result import do_notation, Ok, Err, Result, Do

def fetch_user(id: int) -> Result[dict, str]:
    if id > 0:
        return Ok({"id": id, "name": "Alice", "email": "alice@example.com"})
    return Err("Invalid user ID")

def validate_email(email: str) -> Result[str, str]:
    return Ok(email) if "@" in email else Err("Invalid email")

def send_notification(user: dict) -> Result[str, str]:
    return Ok(f"Notification sent to {user['name']}")

@do_notation
def process_user(user_id: int) -> Do[str, str]:
    user = yield fetch_user(user_id)
    email = yield validate_email(user["email"])
    result = yield send_notification(user)
    return f"{result} at {email}"

print(process_user(1))   # Ok('Notification sent to Alice at alice@example.com')
print(process_user(-1))  # Err('Invalid user ID')

# With exception catching
@do_notation(catch=ValueError)
def parse_and_double(s: str) -> Do[int, str | ValueError]:
    value = yield Ok(int(s))  # ValueError caught and wrapped in Err
    return value * 2

print(parse_and_double("5"))    # Ok(10)
print(parse_and_double("bad"))  # Err(ValueError(...))
```

## do_notation_async - Async Do-Notation

The async version of do-notation for working with async Result-returning functions. Note: async generators must yield the final Ok value.

```python
from result import do_notation_async, Ok, Err, Result, DoAsync
import asyncio

async def fetch_data(url: str) -> Result[dict, str]:
    await asyncio.sleep(0.1)  # Simulate network delay
    if "valid" in url:
        return Ok({"data": f"content from {url}"})
    return Err(f"Failed to fetch {url}")

async def process_data(data: dict) -> Result[str, str]:
    return Ok(data["data"].upper())

@do_notation_async
async def fetch_and_process(url: str) -> DoAsync[str, str]:
    data = yield await fetch_data(url)
    processed = yield await process_data(data)
    yield Ok(f"Processed: {processed}")  # Must yield final Ok

async def main():
    result = await fetch_and_process("https://valid.example.com")
    print(result)  # Ok('Processed: CONTENT FROM HTTPS://VALID.EXAMPLE.COM')

    result = await fetch_and_process("https://bad.example.com")
    print(result)  # Err('Failed to fetch https://bad.example.com')

asyncio.run(main())
```

## combine and partition

`combine` collects multiple Results into a single Result containing a list (all-or-nothing). `partition` separates Results into two lists without short-circuiting.

```python
from result import Ok, Err, combine, partition

results = [Ok(1), Ok(2), Ok(3)]
print(combine(results))  # Ok([1, 2, 3])

results_with_error = [Ok(1), Err("fail"), Ok(3)]
print(combine(results_with_error))  # Err('fail')

# Partition collects all values and errors separately
results = [Ok(1), Err("error1"), Ok(2), Err("error2")]
values, errors = partition(results)
print(values)  # [1, 2]
print(errors)  # ['error1', 'error2']
```

## validate - Applicative Error Accumulation

Unlike monadic chaining, `validate` collects ALL errors instead of short-circuiting on the first one. Ideal for form validation.

```python
from result import Ok, Err, validate, Result

def check_name(name: str) -> Result[str, str]:
    return Ok(name) if len(name) >= 2 else Err("Name too short")

def check_age(age: int) -> Result[int, str]:
    return Ok(age) if age >= 18 else Err("Must be 18+")

def check_email(email: str) -> Result[str, str]:
    return Ok(email) if "@" in email else Err("Invalid email")

# All validations pass
result = validate(
    check_name("Alice"),
    check_age(25),
    check_email("alice@example.com")
)
print(result)  # Ok(('Alice', 25, 'alice@example.com'))

# Accumulates ALL errors
result = validate(
    check_name("A"),
    check_age(16),
    check_email("invalid")
)
print(result)  # Err(['Name too short', 'Must be 18+', 'Invalid email'])
```

## traverse and try_fold

`traverse` maps a fallible function over an iterable, short-circuiting on error. `try_fold` performs a fallible reduction.

```python
from result import Ok, Err, Result, traverse, try_fold

def parse_int(s: str) -> Result[int, str]:
    try:
        return Ok(int(s))
    except ValueError:
        return Err(f"Invalid: {s}")

# traverse: map and collect, short-circuit on error
result = traverse(["1", "2", "3"], parse_int)
print(result)  # Ok([1, 2, 3])

result = traverse(["1", "bad", "3"], parse_int)
print(result)  # Err('Invalid: bad')

# try_fold: fallible accumulation
def safe_divide(acc: float, n: int) -> Result[float, str]:
    if n == 0:
        return Err("Division by zero")
    return Ok(acc / n)

result = try_fold([2, 4, 5], 100.0, safe_divide)
print(result)  # Ok(2.5) -> 100/2=50, 50/4=12.5, 12.5/5=2.5

result = try_fold([2, 0, 5], 100.0, safe_divide)
print(result)  # Err('Division by zero')
```

## flow - Pipeline Orchestrator

Sequential pipeline that chains fallible functions. More readable than multiple `.and_then()` calls.

```python
from result import Ok, Err, Result, flow

def parse(s: str) -> Result[int, str]:
    try:
        return Ok(int(s))
    except ValueError:
        return Err("Parse error")

def validate(n: int) -> Result[int, str]:
    return Ok(n) if n > 0 else Err("Must be positive")

def double(n: int) -> Result[int, str]:
    return Ok(n * 2)

# Clean pipeline
result = flow("10", parse, validate, double)
print(result)  # Ok(20)

result = flow("-5", parse, validate, double)
print(result)  # Err('Must be positive')
```

## ensure and add_context

`ensure` lifts boolean conditions into Results. `add_context` enriches error messages with additional information.

```python
from result import Ok, Err, ensure, add_context, do_notation, Do

@do_notation
def process_order(quantity: int, price: float) -> Do[float, str]:
    yield ensure(quantity > 0, "Quantity must be positive")
    yield ensure(price >= 0, "Price cannot be negative")
    total = quantity * price
    yield ensure(total <= 10000, "Order exceeds maximum")
    return total

print(process_order(5, 100.0))   # Ok(500.0)
print(process_order(-1, 100.0))  # Err('Quantity must be positive')

# Add context to errors
result = Err("connection refused")
enriched = add_context(result, "Database")
print(enriched)  # Err('Database: connection refused')
```

## Async Combinators - gather_results and traverse_async

Concurrent operations for async Result-returning functions with proper error handling.

```python
from result import Ok, Err, Result, gather_results, traverse_async, validate_async
import asyncio

async def fetch_user(id: int) -> Result[dict, str]:
    await asyncio.sleep(0.1)
    if id > 0:
        return Ok({"id": id, "name": f"User{id}"})
    return Err(f"Invalid ID: {id}")

async def main():
    # gather_results: all-or-nothing concurrent execution
    result = await gather_results(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3)
    )
    print(result)  # Ok([{'id': 1, 'name': 'User1'}, ...])

    # First error wins
    result = await gather_results(
        fetch_user(1),
        fetch_user(-1),
        fetch_user(2)
    )
    print(result)  # Err('Invalid ID: -1')

    # validate_async: accumulate ALL errors
    result = await validate_async(
        fetch_user(-1),
        fetch_user(-2)
    )
    print(result)  # Err(['Invalid ID: -1', 'Invalid ID: -2'])

    # traverse_async with concurrency limit
    ids = [1, 2, 3, 4, 5]
    result = await traverse_async(ids, fetch_user, limit=2)
    print(result)  # Ok([...5 user dicts...])

asyncio.run(main())
```

## Outcome Type - Fault-Tolerant Product Type

`Outcome` holds both a value AND an error state simultaneously. Unlike Result (Sum Type), Outcome always contains a value even if errors occurred - ideal for compilers, parsers, and diagnostics.

```python
from result import Outcome, catch_outcome

# Create outcomes with value and optional error
clean = Outcome(42, None)
with_warning = Outcome(42, "Non-critical warning")

print(clean.has_error())        # False
print(with_warning.has_error()) # True

# Native tuple unpacking (Go/Odin style)
value, error = with_warning
print(f"Value: {value}, Error: {error}")  # Value: 42, Error: Non-critical warning

# Convert to strict Result
print(clean.to_result())        # Ok(42)
print(with_warning.to_result()) # Err('Non-critical warning')

# Transform value while preserving error
result = with_warning.map(lambda x: x * 2)
print(result)  # Outcome(value=84, error='Non-critical warning')

# Accumulate errors
out = Outcome(10, None).push_err("e1").push_err("e2")
print(out)  # Outcome(value=10, error=['e1', 'e2'])
```

## catch_outcome Decorator

Catch exceptions into fault-tolerant Outcomes with a default value.

```python
from result import Outcome, catch_outcome
from enum import StrEnum

class ErrorCode(StrEnum):
    PARSE_ERROR = "parse_error"

# Basic usage with default value
@catch_outcome(ValueError, default=0)
def parse_or_default(s: str) -> int:
    return int(s)

print(parse_or_default("42"))   # Outcome(value=42, error=None)
print(parse_or_default("bad"))  # Outcome(value=0, error=ValueError(...))

# Map exception to domain error
@catch_outcome(ValueError, default=-1, map_to=ErrorCode.PARSE_ERROR)
def strict_parse(s: str) -> int:
    return int(s)

print(strict_parse("bad"))  # Outcome(value=-1, error=<ErrorCode.PARSE_ERROR>)
```

## Outcome Chaining and Merging

Chain Outcome operations that accumulate all errors along the way.

```python
from result import Outcome

def step1(x: int) -> Outcome[int, str]:
    return Outcome(x + 10, "step1 warning" if x < 0 else None)

def step2(x: int) -> Outcome[int, str]:
    return Outcome(x * 2, "step2 warning" if x > 50 else None)

# and_then accumulates errors from both outcomes
result = Outcome(5, None).and_then(step1).and_then(step2)
print(result)  # Outcome(value=30, error=None)

result = Outcome(-5, "initial").and_then(step1).and_then(step2)
print(result)  # Outcome(value=10, error=['initial', 'step1 warning'])

# Merge independent outcomes
out1 = Outcome("parsed", "warning1")
out2 = Outcome(42, "warning2")
merged = out1.merge(out2)
print(merged)  # Outcome(value=('parsed', 42), error=['warning1', 'warning2'])
```

## unsafe Namespace for Panicking Operations

The `.unsafe` namespace isolates operations that can raise exceptions, making panics explicit and visible.

```python
from result import Ok, Err, UnwrapError

ok_result = Ok(42)
err_result = Err("something went wrong")

# Safe access (no exceptions)
print(ok_result.ok())           # 42
print(ok_result.err())          # None
print(err_result.unwrap_or(0))  # 0
print(err_result.unwrap_or_else(lambda e: len(e)))  # 20

# Unsafe operations (can panic) - must use .unsafe namespace
print(ok_result.unsafe.unwrap())      # 42
print(err_result.unsafe.unwrap_err()) # 'something went wrong'

# Custom panic message
print(ok_result.unsafe.expect("Should have value"))  # 42

try:
    err_result.unsafe.unwrap()
except UnwrapError as e:
    print(f"Panic: {e}")  # Panic: Called unwrap on an Err value: 'something went wrong'

# Raise custom exception
try:
    err_result.unsafe.unwrap_or_raise(RuntimeError)
except RuntimeError as e:
    print(f"Custom error: {e}")  # Custom error: something went wrong
```

## Type Utilities - from_optional, map2, any_ok

Helper functions for common Result operations.

```python
from result import Ok, Err, from_optional, map2, any_ok

# from_optional: convert Optional to Result
result = from_optional(42, "No value")
print(result)  # Ok(42)

result = from_optional(None, "No value")
print(result)  # Err('No value')

# map2: apply binary function to two Results
result = map2(Ok(10), Ok(5), lambda x, y: x + y)
print(result)  # Ok(15)

result = map2(Ok(10), Err("fail"), lambda x, y: x + y)
print(result)  # Err('fail')

# any_ok: return first success or all errors
result = any_ok([Err("e1"), Ok(42), Err("e2")])
print(result)  # Ok(42)

result = any_ok([Err("e1"), Err("e2"), Err("e3")])
print(result)  # Err(['e1', 'e2', 'e3'])
```

## Summary

Result Pattern provides a comprehensive toolkit for implementing the "Errors as Values" pattern in Python. The primary use cases include: building type-safe APIs where error handling is explicit in function signatures; form and data validation with error accumulation using `validate`; building compilers or parsers with the fault-tolerant `Outcome` type that preserves partial results; concurrent async workflows with `gather_results` and `traverse_async`; and progressive migration from exception-based code using `@catch` decorators.

Integration patterns typically involve using `@catch` at system boundaries to lift exception-throwing library code into Result types, using `@do_notation` for complex business logic with multiple fallible steps, employing `validate` for user input validation where all errors should be reported, and using `Outcome` for batch processing or compilation pipelines where partial success is valuable. The `.unsafe` namespace provides an escape hatch for trusted code paths while keeping panics explicit and visible in the codebase.