okresult 0.4.0

A lightweight Result type for Python.

okresult

Lightweight Result type for Python, inspired by better-result.

Install

pip install okresult

Quick Start

from okresult import Result, safe, fn
import json

# Wrap throwing functions
def load_user() -> dict[str, str]:
    return json.loads('{"name": "John", "age": 30}')
parsed = safe(load_user)

# Check and use
if parsed.is_ok():
    print(parsed.unwrap())
else:
    print(parsed.unwrap_err())

# Or use pattern matching
message = parsed.match({
    "ok": fn[dict[str, str], str](lambda data: f"Got: {data['name']}"),
    "err": fn[object, str](lambda e: f"Failed: {e.cause}"),
})

Contents

• Creating Results
• Transforming Results
• Handling Errors
• Extracting Values
• Generator Composition
• Panic
• Retry Support
• Tagged Errors
• Serialization
• Typed Lambda Expressions
• API Reference

Creating Results

from okresult import Result, Ok, Err, safe, safe_async, fn

# Success
ok = Result.ok(42)

# Error
err = Result.err(ValueError("failed"))

# From throwing function
def risky() -> float:
    raise ValueError("Invalid input")

result = safe(risky)

# From async function
async def risky_async() -> float:
    raise ValueError("Invalid input")

result = await safe_async(risky_async)

# With custom error handling
result = safe({"try_": risky, "catch": fn[Exception, str](lambda e: "Error: " + str(e))})

Transforming Results

from okresult import Ok, Err, map as result_map, fn

result = (
    Ok[int, ValueError](2)
    .map(fn[int, int](lambda x: x * 2))  # Ok(4)
    .and_then(
        # Chain Result-returning functions
        fn[int, Result[int, ValueError]](lambda x: Ok[int, ValueError](x) if x > 0 else Err[int, ValueError](ValueError("negative")))
    )
)

# Standalone functions (data-first or data-last)
result_map(result, fn[int, int](lambda x: x + 1))
result_map(fn[int, int](lambda x: x + 1))(result)  # Pipeable

Handling Errors

from okresult import Result, TaggedError, fn
from typing import Union, TypeAlias

err_result: Result[int, ValueError] = Result.err(ValueError("invalid"))

# Transform errors
err_result.map_err(fn[ValueError, RuntimeError](lambda e: RuntimeError(str(e))))  # Err(RuntimeError(...))


# Recover from specific errors
class NotFoundError(TaggedError):
    TAG = "NotFoundError"

    __slots__ = ("id",)
    
    def __init__(self, id: str) -> None:
        super().__init__(f"Not found: {id}")
        self.id = id

class ValidationError(TaggedError):
    TAG = "ValidationError"
    __slots__ = ("field",)
    
    def __init__(self, field: str) -> None:
        super().__init__(f"Invalid: {field}")
        self.field = field

AppError: TypeAlias = Union[NotFoundError, ValidationError]

def fetch_user(id: str) -> Result[dict[str, str], AppError]:
    if id == "valid":
        return Result.ok({"name": "John", "id": id})
    if not id:
        return Result.err(ValidationError("id"))
    return Result.err(NotFoundError(id))

default_user = {"name": "Default User"}


result = fetch_user("123").match({
    "ok": fn[dict[str, str], Result[dict[str, str], AppError]](lambda user: Result.ok(user)),
    "err": fn[AppError, Result[dict[str, str], AppError]](
        lambda e: Result.ok(default_user) if e.tag == "NotFoundError" else Result.err(e)
    )
})

Extracting Values

from okresult import Result, unwrap, fn

result_ok = Result.ok(42)
result_err = Result.err(ValueError("invalid"))

# Unwrap (throws on Err)
value = unwrap(result_ok)
value = result_ok.unwrap()

# With fallback
value = result_err.unwrap_or(0)

# Pattern match
value = result_err.match({
    "ok": fn[int, int](lambda v: v),
    "err": fn[ValueError, int](lambda e: 0),
})

Generator Composition

Chain multiple Results without nested callbacks or early returns. Values are automatically unwrapped on success and short-circuited on error.

Why?

Without generator composition, chaining many operations leads to deeply nested callbacks or "callback hell":

# Without generator composition - deeply nested callbacks
def parse_number(s: str) -> Result[int, str]:
    try:
        return Result.ok(int(s))
    except ValueError:
        return Result.err(f"Invalid number: {s}")

def divide(a: int, b: int) -> Result[float, str]:
    if b == 0:
        return Result.err("Division by zero")
    return Result.ok(a / b)

result = parse_number("10").and_then(
    lambda a: parse_number("2").and_then(
        lambda b: divide(a, b).and_then(
            lambda c: Result.ok(c * 2)  # Another operation
        )
    )
)

With generator composition, the same code reads linearly:

# With generator composition - clean and readable
def compute() -> Do[float, str]:
    a: int = yield parse_number("10")
    b: int = yield parse_number("2")
    c: float = yield divide(a, b)
    d: float = yield Result.ok(c * 2)  # Another operation
    return Result.ok(d)

result = Result.gen(compute) 
# ^Result.ok(10)

Generator composition scales infinitely - you can chain as many operations as needed without nesting getting deeper. Each yield automatically unwraps Ok values and short-circuits on Err.

Synchronous Composition

from okresult import Result, Do

def parse_number(s: str) -> Result[int, str]:
    try:
        return Result.ok(int(s))
    except ValueError:
        return Result.err(f"Invalid number: {s}")

def divide(a: int, b: int) -> Result[float, str]:
    if b == 0:
        return Result.err("Division by zero")
    return Result.ok(a / b)

def compute() -> Do[float, str]:
    a: int = yield parse_number("10")  # Unwraps or short-circuits
    b: int = yield parse_number("2")
    c: float = yield divide(a, b)
    return Result.ok(c)

result = Result.gen(compute)
# Result[float, str] - errors from all yields are collected into union type

# Short-circuiting on error
def compute_with_error() -> Do[float, str]:
    a: int = yield parse_number("10")
    b: int = yield parse_number("invalid")  # This will short-circuit
    c: float = yield divide(a, b)
    return Result.ok(c)

result_err = Result.gen(compute_with_error)
# Returns Err("Invalid number: invalid")

Async Composition

from okresult import Result, DoAsync
import asyncio

async def fetch_user(id: int) -> Result[dict[str, str], str]:
    await asyncio.sleep(0.001)  # Simulate async work
    if id > 0:
        return Result.ok({"name": "John", "id": str(id)})
    return Result.err("Invalid user ID")

async def fetch_posts(user_id: str) -> Result[list[dict[str, str]], str]:
    await asyncio.sleep(0.001)  # Simulate async work
    return Result.ok([{"title": "Post 1"}, {"title": "Post 2"}])

async def load_user_data() -> DoAsync[dict[str, object], str]:
    user_id = 123
    user = yield await fetch_user(user_id)
    posts = yield await fetch_posts(user["id"])
    yield Result.ok({"user": user, "posts": posts})

result = await Result.gen_async(load_user_data)
# Result[dict[str, object], str]

Errors from all yielded Results are automatically collected into the final error union type. The generator short-circuits on the first Err, so subsequent yields are skipped.

Note: Async generators must yield the final Result as the last value (non-empty return statements are a SyntaxError per PEP 525). Raising StopAsyncIteration with a value is not supported at the moment by choice. Always use yield Result.ok(...) instead.

Panic

Thrown (not returned) when user callbacks throw inside Result operations. Represents a defect in your code, not a domain error.

from okresult import Result, Panic, fn

# Callback throws → Panic
try:
    Result.ok(1).map(fn[int, int](lambda x: 1 // 0))
except Panic as p:
    print(f"Panic: {p.message}, caused by: {p.cause}")

# Catch handler throws → Panic  
try:
    safe({
        "try_": lambda: 1 / 1,
        "catch": fn[Exception, int](lambda e: 1 // 0)  # Bug in handler
    })
except Panic as p:
    print(f"Panic: {p.message}, caused by: {p.cause}")

# and_then callback throws → Panic
try:
    Result.ok(1).and_then(fn[int, Result[int, str]](lambda x: 1 // 0))
except Panic as p:
    print(f"Panic: {p.message}, caused by: {p.cause}")

# match handler throws → Panic
try:
    Result.ok(1).match({
        "ok": fn[int, int](lambda x: 1 // 0),  # Bug in handler
        "err": fn[str, int](lambda e: 0)
    })
except Panic as p:
    print(f"Panic: {p.message}, caused by: {p.cause}")

Why Panic?

Err is for recoverable domain errors. Panic is for bugs — like Rust's panic!(). If your .map() callback throws, that's not an error to handle, it's a defect to fix. Returning Err would collapse type safety (Result[T, E] becomes Result[T, E | Unknown]).

Panic Properties

Property	Type	Description
message	str	Describes where/what panicked (e.g., "map failed")
cause	Exception	The exception that was thrown

Retry Support

from okresult import safe, safe_async, fn

def risky() -> float:
    raise ValueError("Invalid input")

# Sync retry
result = safe(risky, {"retry": {"times": 3}})

# Async retry with backoff
async def fetch(url: str) -> str:
    raise ConnectionError("Network error")

result = await safe_async(
    lambda: fetch("https://api.example.com"),  
    {
        "retry": {
            "times": 3,
            "delay_ms": 100,
            "backoff": "exponential",  # or "linear" | "constant"
            # Optional: only retry for certain error types/flags
            "should_retry": lambda e: isinstance(e, ConnectionError),
        },
    },
)

# With custom error mapping and richer predicates
from okresult import TaggedError
from typing import Union

class RetryableException(TaggedError):
    TAG = "RetryableException"
    __slots__ = ("msg",)
    
    def __init__(self, msg: str) -> None:
        super().__init__(msg)
        self.msg = msg

class NonRetryableException(TaggedError):
    TAG = "NonRetryableException"
    __slots__ = ("msg",)
    
    def __init__(self, msg: str) -> None:
        super().__init__(msg)
        self.msg = msg

ApiError = Union[RetryableException, NonRetryableException]

async def call_api(url: str) -> str:
    raise RuntimeError("rate limited")

result = await safe_async(
    {
        "try_": lambda: call_api("https://api.example.com"),
        "catch": fn[Exception, ApiError](
            lambda e: RetryableException(str(e)) if "rate limited" in str(e) else NonRetryableException(str(e))
        ),
    },
    {
        "retry": {
            "times": 3,
            "delay_ms": 100,
            "backoff": "exponential",
            "should_retry": lambda e: e.tag == "RetryableException",
        },
    },
)

# Async predicates: For async decisions (e.g. feature flags, rate limits)
async def check_rate_limit() -> bool:
    return True

result = await safe_async(
    lambda: call_api("https://api.example.com"),
    {
        "retry": {
            "times": 3,
            "delay_ms": 100,
            "backoff": "exponential",
            # should_retry can be async in safe_async
            "should_retry": async lambda e: await check_rate_limit(),
        },
    },
)

UnhandledException

When safe() or safe_async() catches an exception without a custom handler, the error type is UnhandledException:

from okresult import Result, UnhandledException, safe, safe_async, TaggedError, fn
import json

# Automatic — error type is UnhandledException
def parse_json(input: str) -> dict[str, str]:
    return json.loads(input)

result = safe(lambda: parse_json('{"key": "value"}'))  

# Custom handler — you control the error type using TaggedError
class ParseError(TaggedError):
    TAG = "ParseError"
    __slots__ = ("cause",)
    
    def __init__(self, cause: Exception) -> None:
        super().__init__(f"Parse failed: {str(cause)}")
        self.cause = cause

result = safe({
    "try_": lambda: parse_json('{"key": "value"}'),  
    "catch": fn[Exception, ParseError](lambda e: ParseError(e))
})

# Same for async
async def fetch_and_parse(json_str: str) -> dict[str, str]:
    # Simulate async work
    return parse_json(json_str)

# Async with custom error handler
result = await safe_async({
    "try_": lambda: fetch_and_parse('invalid'),  
    "catch": fn[Exception, ParseError](lambda e: ParseError(e))
})

Tagged Errors

from okresult import Result, TaggedError, fn
from typing import Union, TypeAlias

class NotFoundError(TaggedError):
    TAG = "NotFoundError"
    __slots__ = ("id",)
    def __init__(self, id: str) -> None:
        super().__init__(f"Not found: {id}")
        self.id = id

class ValidationError(TaggedError):
    TAG = "ValidationError"
    __slots__ = ("field",)
    def __init__(self, field: str) -> None:
        super().__init__(f"Invalid: {field}")
        self.field = field

AppError: TypeAlias = Union[NotFoundError, ValidationError] 

result_err = Result.err(ValidationError("name"))

# Exhaustive matching
result_exhaustive = TaggedError.match(
    result_err.unwrap_err(),
    {
        ValidationError: fn[ValidationError, Result[dict[str, str], ValidationError]](lambda e: Result.ok({"message": f"Invalid: {e.field}"})),
        NotFoundError: fn[NotFoundError, Result[dict[str, str], ValidationError]](lambda e: Result.ok({"name": "Default User"})),
    }
)

# Partial matching with a fallback
result_partial = TaggedError.match_partial(
    result_err.unwrap_err(),
    {
        ValidationError: fn[ValidationError, Result[dict[str, str], ValidationError]](lambda e: Result.ok({"message": f"Invalid: {e.field}"})),
        NotFoundError: fn[NotFoundError, Result[dict[str, str], ValidationError]](lambda e: Result.ok({"name": "Default User"})),
    },
    otherwise=fn[TaggedError, Result[dict[str, str], ValidationError]](lambda e: Result.ok({"message": "Unknown error"}))
)

Serialization

Rehydrate Results from JSON for storage or network transfer.

from okresult import Result, fn
import json

# Serialize a Result to JSON (e.g., storage or network transfer)
original = Result.ok(42)
serialized_dict = original.serialize()            # {'status': 'ok', 'value': 42}
serialized_json = json.dumps(serialized_dict)     # "{\"status\":\"ok\",\"value\":42}"

# Rehydrate the serialized Result back to a Result instance
hydrated = Result.hydrate(json.loads(serialized_json))


# Now you can use Result methods again
doubled = hydrated.map(fn[int, int](lambda x: x * 2))  # Ok(84)

# Works with Err too
err_result = Result.err(ValueError("failed"))
err_json = json.dumps(err_result.serialize())     # "{\"status\":\"err\",\"value\":\"failed\"}"
rehydrated = Result.hydrate(json.loads(err_json))

# Note: Exceptions are serialized as strings for portability.
# Rehydrating an Err produced from an Exception yields Err("failed") (a string),
# not an Exception instance. Use typed hydration to reconstruct specific types.

# Typed hydration with decoders
def decode_int(x: object) -> int:
    if isinstance(x, int):
        return x
    raise ValueError("expected int")

def decode_error(x: object) -> ValueError:
    # Turn the serialized error payload back into a ValueError
    return ValueError(str(x))

typed: Result[int, ValueError] | None = Result.hydrate_as(
    json.loads(serialized_json),
    ok=decode_int,
    err=decode_error,
)

Typed Lambda Expressions

Python doesn't support type annotations inside lambda expressions, making it difficult to express precise types for inline functions. The fn helper provides a way to create typed callables using subscript syntax, achieving a close semantic equivalent to a "typed lambda" in Python.

from okresult import fn

# Explicitly typed lambda
double = fn[int, int](lambda x: x * 2)
add_one = fn[int, int](lambda x: x + 1)

# Works with Result transformations
result = Result.ok(5).map(double)  # Ok(10)

# Works with match handlers
message = result.match({
    "ok": fn[int, str](lambda x: f"Value: {x}"),
    "err": fn[str, str](lambda e: f"Error: {e}"),
})

At runtime, fn returns the provided callable unchanged (no wrapping, allocation, or indirection). All type information is enforced statically by type checkers (e.g., pyright, mypy). The above double example is equivalent, from a typing perspective, to defining a named function:

def double(x: int) -> int:
    return x * 2

Instead of requiring verbose named functions or explicit Callable annotations, fn shifts the type declaration to the call boundary while keeping lambdas concise and idiomatic. It's particularly useful in functional-style APIs (e.g., map, and_then, match) where lambdas are common and precise typing significantly improves developer experience.

API Reference

Types

Type	Description
Result[A, E]	Base type for results (Ok or Err)
Ok[A, E]	Success variant
Err[A, E]	Error variant
Do[A, E]	Type alias for Generator[Result[Any, E], Any, Result[A, E]]
DoAsync[A, E]	Type alias for AsyncGenerator[Result[Any, E], Any]
Matcher[A, B, E, F]	TypedDict for pattern matching
TaggedError	Base class for tagged errors
UnhandledException	Error type for unhandled exceptions

Result Creation

Function	Description
Result.ok(value)	Create success result
Result.err(error)	Create error result
Result.gen(fn, context?)	Generator-based Result composition (do-notation)
Result.gen_async(fn, context?)	Async generator-based Result composition
Result.flatten(result)	Flatten nested Result into single Result
Result.partition(results)	Partition iterable of Results into a tuppl (list[ok_values], list[err_values])
Result.hydrate(data)	Deserialize from dict
Result.hydrate_as(data, *, ok, err)	Typed deserialization with decoders
Ok(value)	Create Ok instance
Err(error)	Create Err instance
safe(fn, config?)	Wrap throwing function with optional retry
safe_async(fn, config?)	Wrap async function with optional retry

Module-Level Functions (Data-First & Data-Last)

Function	Description
map(result, fn) or map(fn)(result)	Transform success value
map_err(result, fn) or map_err(fn)(result)	Transform error value
tap(result, fn) or tap(fn)(result)	Side effect on success
tap_async(result, fn) or tap_async(fn)(result)	Async side effect on success
and_then(result, fn) or and_then(fn)(result)	Chain Result-returning function
and_then_async(result, fn) or and_then_async(fn)(result)	Chain async Result-returning function
match(result, handlers) or match(handlers)(result)	Pattern match on Result
unwrap(result, message?)	Extract value or raise

Instance Methods

Method	Description
.status	Property: "ok" or "err"
.is_ok()	Check if Ok
.is_err()	Check if Err
.map(fn)	Transform success value
.map_err(fn)	Transform error value
.serialize()	Serialize to dict for storage/transport
.and_then(fn)	Chain Result-returning function
.and_then_async(fn)	Chain async Result-returning function
.match({"ok": fn, "err": fn})	Pattern match
.unwrap(message?)	Extract value or raise
.unwrap_or(fallback)	Extract value or return fallback
.unwrap_err(message?)	Extract error or raise
.tap(fn)	Side effect on success
.tap_async(fn)	Async side effect on success

TaggedError Methods

Method	Description
TaggedError.is_error(value)	Type guard for Exception instances
TaggedError.is_tagged_error(value)	Type guard for TaggedError instances
TaggedError.match(error, handlers)	Exhaustive match by tag string
TaggedError.match_partial(error, handlers, otherwise)	Partial match by tag string with fallback
.tag	Property: error tag string
.message	Property: error message

Configuration Types

Type	Description
SafeConfig	Configuration for safe()
SafeConfigAsync	Configuration for safe_async()
SafeOptions[A, E]	Options with custom error mapping
RetryConfig	Retry configuration for sync operations
RetryConfigAsync	Retry configuration for async operations

License

MIT