Result

The strict Sum Type for functional error handling.

result.result

Result: Functional Error Handling for Modern Python.

A lightweight, single-file library designed to implement the 'Errors as Values' pattern in Python 3.14+. This library tries to help bridge the gap between pure functional safety and the pragmatic realities of the exception-heavy Python ecosystem.

Core Philosophy

In standard Python, errors are implicit side-effects (Exceptions). In this library, errors are explicit return values. This forcing function ensures that failure paths are handled as diligently as success paths.

Core Goals

1. Explicit Error Handling: Shift from 'invisible' exceptions to explicit Result return types, making failure modes a first-class part of the API.
2. Reduced Cognitive Load: Use the @do and @do_async decorators to write linear, procedural-looking code that automatically handles short-circuiting logic.
3. Static Analysis First: Leverage modern Python typing features (PEP 695, PEP 742) to provide perfect type narrowing and static verification in tools like Basedpyright, Mypy and Ty.
4. Zero-Escape Safety: Isolate crashing operations (panics) within the .unsafe namespace to ensure that 'unwrapping' is always an intentional, visible choice.
5. Pragmatic Interop: Provide 'lifting' tools like @catch to seamlessly convert standard Python exception-throwing code into functional containers.

Limitations & Constraints

• Type Invariance: Python's type system can be rigid regarding variance in Unions. Returning a specific Err(ValueError) when a signature expects Result[T, Exception] may occasionally require explicit type annotations.
• Higher-Kinded Types: Python lacks native support for higher-kinded types, making operations like flatten() return Result[Any, Any] to maintain runtime flexibility while sacrificing some static detail.
• Ecosystem Inertia: External Python libraries will continue to raise exceptions. Lifting must be performed at the integration boundaries.

OkErr module-attribute

OkErr: Final = (Ok, Err)

A constant for use in isinstance checks.

Examples:

>>> res = Ok(10)
>>> isinstance(res, OkErr)
True

Result module-attribute

Result = Union['Ok[T_co]', 'Err[E_co]']

A type that represents either success (Ok) or failure (Err).

This is the foundational type for explicit error handling. Use is_ok() and is_err() to narrow the type or match for exhaustive handling.

Examples:

>>> def get_user(id: int) -> Result[dict, str]:
...     return Ok({"id": id}) if id > 0 else Err("Invalid ID")

Do

Do = Generator[Result[Any, E], Any, T]

A type alias for generator functions compatible with the @do decorator.

The generator yields Result variants to unwrap them and eventually returns a value of type T.

DoAsync

DoAsync = AsyncGenerator[
    Result[T, E] | Result[Any, E], Any
]

A type alias for async generator functions compatible with the @do_async decorator.

Async generators must yield an Ok variant as their final step to simulate a return value.

AssertOk

AssertOk(message: str = 'Result was Err')

A context manager for asserting that Results must be Ok.

It automatically monitors local variable assignments within the block. If any local variable is assigned an Err variant, it raises an AssertionError immediately (fail-fast).

Note

The automatic scanning only works for the local scope where the with assert_ok() block is defined.

Initialize the assert_ok context with a custom message.

Source code in src/result/result.py

def __init__(self, message: str = "Result was Err") -> None:
    """Initialize the assert_ok context with a custom message."""
    self.message = message
    self._initial_locals: set[str] = set()
    self._old_trace: Any = None
    self._is_scanning: bool = False

__enter__

__enter__() -> AssertOk

Enter the assert_ok context and install the fail-fast tracer.

Source code in src/result/result.py

def __enter__(self) -> AssertOk:
    """Enter the assert_ok context and install the fail-fast tracer."""
    # Capture current locals to avoid re-triggering on existing variables
    frame = sys._getframe(1)
    self._initial_locals = set(frame.f_locals.keys())

    # Install trace function for fail-fast detection
    self._old_trace = sys.gettrace()
    sys.settrace(self._trace_callback)
    return self

__exit__

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Exit the assert_ok context and uninstall the tracer.

Source code in src/result/result.py

def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
    """Exit the assert_ok context and uninstall the tracer."""
    sys.settrace(self._old_trace)

check

check(result: Result[T, E]) -> T

Verify that a result is Ok within the context.

Parameters:

Name	Type	Description	Default
result	Result[T, E]	The Result to verify.	required

Returns:

Type	Description
T	The success value if Ok.

Raises:

Type	Description
AssertionError	If the result is an Err.

Source code in src/result/result.py

def check[T, E](self, result: Result[T, E]) -> T:
    """Verify that a result is Ok within the context.

    Args:
        result: The Result to verify.

    Returns:
        The success value if Ok.

    Raises:
        AssertionError: If the result is an Err.

    """
    __tracebackhide__ = True
    match result:
        case Err(e):  # pyright: ignore[reportUnknownVariableType]
            err_val = cast("Any", e)
            return _raise_assertion_error(f"{self.message}: {err_val}")
        case Ok(v):
            return v

CatchContext

CatchContext(exceptions: tuple[type[Exception], ...])

A context manager for capturing exceptions and converting them to Results.

Usage

with catch(ValueError) as ctx: ... ctx.set(int("123")) ctx.result Ok(123)

Source code in src/result/result.py

def __init__(self, exceptions: tuple[type[Exception], ...]) -> None:
    self.exceptions = exceptions
    self.result: Result[T, E] | None = None

set

set(value: T) -> None

Set the success value for the context.

Source code in src/result/result.py

def set(self, value: T) -> None:
    """Set the success value for the context."""
    self.result = Ok(value)

Err dataclass

Err(error: E_co)

A failed result containing an error state.

To handle the error safely, use pattern matching or recovery methods like .or_else() or .unwrap_or().

Examples:

>>> res = Err("not found")
>>> match res:
...     case Err(e):
...         print(f"Error: {e}")
Error: not found

Initialize an Err variant.

Parameters:

Name	Type	Description	Default
error	E_co	The error state or exception to wrap.	required

Source code in src/result/result.py

def __init__(self, error: E_co) -> None:
    """Initialize an Err variant.

    Args:
        error: The error state or exception to wrap.

    """
    object.__setattr__(self, "_error", error)

error property

error: E_co

Access the error state.

Note: While accessible on the Err variant, it is recommended to use pattern matching or functional methods for maximum safety.

unsafe property

unsafe: _ErrUnsafe[E_co]

Namespace for operations that might panic.

Examples:

>>> Err("fail").unsafe.unwrap_err()
'fail'

__add__

__add__(other: Any) -> Result[Any, Any]

Support short-circuiting addition (+) operator.

Always returns self for Err variants, short-circuiting the addition.

Source code in src/result/result.py

def __add__(self, other: Any) -> Result[Any, Any]:
    """Support short-circuiting addition (+) operator.

    Always returns self for Err variants, short-circuiting the addition.
    """
    if isinstance(other, (Ok, Err)):
        return self
    return NotImplemented

__aiter__ async

__aiter__() -> AsyncGenerator[Never, Any]

Allow use in async generator expressions for do-notation.

Source code in src/result/result.py

async def __aiter__(self) -> AsyncGenerator[Never, Any]:
    """Allow use in async generator expressions for do-notation."""
    raise _DoError(self)
    yield self  # should be UNREACHABLE

__bool__

__bool__() -> Literal[False]

Truthiness check. Returns False for Err.

This enables idiomatic Python conditional narrowing: if res: # type is narrowed to Ok

Source code in src/result/result.py

def __bool__(self) -> Literal[False]:
    """Truthiness check. Returns False for Err.

    This enables idiomatic Python conditional narrowing:
    `if res: # type is narrowed to Ok`
    """
    return False

__getattr__

__getattr__(name: str) -> Never

Educational runtime safeguard against common API mistakes.

Source code in src/result/result.py

def __getattr__(self, name: str) -> Never:
    """Educational runtime safeguard against common API mistakes."""
    if name in {
        "value",
        "error",
        "unwrap",
        "unwrap_err",
        "expect",
        "expect_err",
        "unwrap_or_raise",
        "inspect",
        "inspect_async",
        "inspect_err",
    }:
        _raise_api_error(name)
    msg = f"'{self.__class__.__name__}' object has no attribute '{name}'"
    raise AttributeError(msg)

__hash__

__hash__() -> int

Explicit hash to prevent collisions with Ok variants.

Source code in src/result/result.py

def __hash__(self) -> int:
    """Explicit hash to prevent collisions with Ok variants."""
    return hash((Err, self._error))

__iter__

__iter__() -> Generator[Result[Any, E_co], Any, Never]

Allow use in generator expressions for do-notation.

Source code in src/result/result.py

def __iter__(self) -> Generator[Result[Any, E_co], Any, Never]:
    """Allow use in generator expressions for do-notation."""
    raise _DoError(self)
    yield self  # should be UNREACHABLE

and_then

and_then(
    _func: Callable[[Any], Result[Any, Any]],
) -> Err[E_co]

Short-circuit the chain and return self unchanged.

Source code in src/result/result.py

def and_then(self, _func: Callable[[Any], Result[Any, Any]]) -> Err[E_co]:
    """Short-circuit the chain and return self unchanged."""
    return self

and_then_async async

and_then_async(
    _func: Callable[[T_local], Awaitable[Result[U, E_co]]],
) -> Err[E_co]

Short-circuit async success chaining.

Parameters:

Name	Type	Description	Default
_func	Callable[[T_local], Awaitable[Result[U, E_co]]]	Ignored async success chaining function.	required

Returns:

Type	Description
Err[E_co]	The current instance unchanged.

Source code in src/result/result.py

async def and_then_async[U, T_local](self, _func: Callable[[T_local], Awaitable[Result[U, E_co]]]) -> Err[E_co]:
    """Short-circuit async success chaining.

    Args:
        _func: Ignored async success chaining function.

    Returns:
        The current instance unchanged.

    """
    return self

cast_types

cast_types() -> Result[U, F]

Zero-runtime-cost type hint override for strict variance edge cases.

This allows manually guiding the type checker when it fails to infer complex union types correctly.

Returns:

Type	Description
Result[U, F]	The same instance, but with new type parameters for the checker.

Source code in src/result/result.py

def cast_types[U, F](self) -> Result[U, F]:
    """Zero-runtime-cost type hint override for strict variance edge cases.

    This allows manually guiding the type checker when it fails to infer
    complex union types correctly.

    Returns:
        The same instance, but with new type parameters for the checker.

    """
    return cast("Result[U, F]", self)

err

err() -> E_co

Convert to Optional[E]. Returns the error state.

Returns:

Type	Description
E_co	The contained error state.

Examples:

>>> Err("fail").err()
'fail'

Source code in src/result/result.py

def err(self) -> E_co:
    """Convert to Optional[E]. Returns the error state.

    Returns:
        The contained error state.

    Examples:
        >>> Err("fail").err()
        'fail'

    """
    return self._error

expect

expect(_msg: str) -> Never

Root-level expect is disabled. Use .unsafe.expect() instead.

Source code in src/result/result.py

def expect(self, _msg: str) -> Never:
    """Root-level expect is disabled. Use .unsafe.expect() instead."""
    _raise_api_error("expect")

expect_err

expect_err(_msg: str) -> Never

Root-level expect_err is disabled. Use .unsafe.expect_err() instead.

Source code in src/result/result.py

def expect_err(self, _msg: str) -> Never:
    """Root-level expect_err is disabled. Use .unsafe.expect_err() instead."""
    _raise_api_error("expect_err")

filter

filter(
    _predicate: Callable[[Any], bool], _error: object
) -> Err[E_co]

Ignore the filter and return self unchanged.

Source code in src/result/result.py

def filter(self, _predicate: Callable[[Any], bool], _error: object) -> Err[E_co]:
    """Ignore the filter and return self unchanged."""
    return self

flatten

flatten() -> Err[E_co]

Short-circuit the flattening and return self unchanged.

Returns:

Type	Description
Err[E_co]	The current instance unchanged.

Source code in src/result/result.py

def flatten(self) -> Err[E_co]:
    """Short-circuit the flattening and return self unchanged.

    Returns:
        The current instance unchanged.

    """
    return self

inspect

inspect(_func: Callable[[Any], Any]) -> Never

Root-level inspect is disabled. Use .tap() instead.

Source code in src/result/result.py

def inspect(self, _func: Callable[[Any], Any]) -> Never:
    """Root-level inspect is disabled. Use .tap() instead."""
    _raise_api_error("inspect")

inspect_async

inspect_async(
    _func: Callable[[Any], Awaitable[Any]],
) -> Never

Root-level inspect_async is disabled. Use .tap_async() instead.

Source code in src/result/result.py

def inspect_async(self, _func: Callable[[Any], Awaitable[Any]]) -> Never:
    """Root-level inspect_async is disabled. Use .tap_async() instead."""
    _raise_api_error("inspect_async")

inspect_err

inspect_err(_func: Callable[[Any], Any]) -> Never

Root-level inspect_err is disabled. Use .tap_err() instead.

Source code in src/result/result.py

def inspect_err(self, _func: Callable[[Any], Any]) -> Never:
    """Root-level inspect_err is disabled. Use .tap_err() instead."""
    _raise_api_error("inspect_err")

is_err

is_err() -> Literal[True]

Check if the result is an Err variant.

Returns:

Type	Description
Literal[True]	Always True for Err instances.

Source code in src/result/result.py

def is_err(self) -> Literal[True]:
    """Check if the result is an Err variant.

    Returns:
        Always True for Err instances.

    """
    return True

is_err_and

is_err_and(predicate: Callable[[E_co], bool]) -> bool

Check if the result is an Err variant and matches a predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[E_co], bool]	A function to test the error state.	required

Returns:

Type	Description
bool	True if the result is Err and the predicate returns True.

Examples:

>>> Err(404).is_err_and(lambda e: e == 404)
True

Source code in src/result/result.py

def is_err_and(self, predicate: Callable[[E_co], bool]) -> bool:
    """Check if the result is an Err variant and matches a predicate.

    Args:
        predicate: A function to test the error state.

    Returns:
        True if the result is Err and the predicate returns True.

    Examples:
        >>> Err(404).is_err_and(lambda e: e == 404)
        True

    """
    return predicate(self._error)

is_ok

is_ok() -> Literal[False]

Check if the result is an Ok variant.

Returns:

Type	Description
Literal[False]	Always False for Err instances.

Source code in src/result/result.py

def is_ok(self) -> Literal[False]:
    """Check if the result is an Ok variant.

    Returns:
        Always False for Err instances.

    """
    return False

is_ok_and

is_ok_and(
    _predicate: Callable[[Any], bool],
) -> Literal[False]

Check if the result is an Ok variant and matches a predicate.

Always returns False for Err instances.

Parameters:

Name	Type	Description	Default
_predicate	Callable[[Any], bool]	Ignored for Err instances.	required

Returns:

Type	Description
Literal[False]	Always False.

Source code in src/result/result.py

def is_ok_and(self, _predicate: Callable[[Any], bool]) -> Literal[False]:
    """Check if the result is an Ok variant and matches a predicate.

    Always returns False for Err instances.

    Args:
        _predicate: Ignored for Err instances.

    Returns:
        Always False.

    """
    return False

map

map(_func: Callable[[T_local], U]) -> Err[E_co]

Ignore the value mapping and return self unchanged.

Source code in src/result/result.py

def map[U, T_local](self, _func: Callable[[T_local], U]) -> Err[E_co]:
    """Ignore the value mapping and return self unchanged."""
    return self

map_async async

map_async(
    _func: Callable[[T_local], Awaitable[U]],
) -> Err[E_co]

Ignore success mapping and return self unchanged.

Parameters:

Name	Type	Description	Default
_func	Callable[[T_local], Awaitable[U]]	Ignored async mapping function.	required

Returns:

Type	Description
Err[E_co]	The current instance unchanged.

Source code in src/result/result.py

async def map_async[U, T_local](self, _func: Callable[[T_local], Awaitable[U]]) -> Err[E_co]:
    """Ignore success mapping and return self unchanged.

    Args:
        _func: Ignored async mapping function.

    Returns:
        The current instance unchanged.

    """
    return self

map_err

map_err(func: Callable[[E_co], F]) -> Err[F]

Apply a function to the contained error.

Parameters:

Name	Type	Description	Default
func	Callable[[E_co], F]	A function to transform the error state.	required

Returns:

Type	Description
Err[F]	A new Err containing the transformed error.

Examples:

>>> Err(404).map_err(lambda code: f"Code: {code}")
Err('Code: 404')

Source code in src/result/result.py

def map_err[F](self, func: Callable[[E_co], F]) -> Err[F]:
    """Apply a function to the contained error.

    Args:
        func: A function to transform the error state.

    Returns:
        A new `Err` containing the transformed error.

    Examples:
        >>> Err(404).map_err(lambda code: f"Code: {code}")
        Err('Code: 404')

    """
    return Err(func(self._error))

map_exc

map_exc(mapping: Mapping[type[Exception], Any]) -> Err[Any]

Replace the error payload if its type exists in the mapping.

This is ideal for converting raw Python exceptions into domain-specific Enums or error codes immediately after they are caught.

Parameters:

Name	Type	Description	Default
mapping	Mapping[type[Exception], Any]	A dictionary mapping exception types to new values.	required

Returns:

Type	Description
Err[Any]	A new Err with the mapped value if a match was found, otherwise self.

Examples:

>>> # Single mapping to Enum
>>> Err(ValueError("fail")).map_exc({ValueError: ErrorCode.INVALID})
Err(<ErrorCode.INVALID: 'invalid'>)

>>> # Multiple mappings
>>> mapping = {ValueError: ErrorCode.INVALID, KeyError: ErrorCode.MISSING}
>>> Err(KeyError("key")).map_exc(mapping)
Err(<ErrorCode.MISSING: 'missing'>)

Source code in src/result/result.py

def map_exc(self, mapping: Mapping[type[Exception], Any]) -> Err[Any]:
    """Replace the error payload if its type exists in the mapping.

    This is ideal for converting raw Python exceptions into domain-specific
    Enums or error codes immediately after they are caught.

    Args:
        mapping: A dictionary mapping exception types to new values.

    Returns:
        A new Err with the mapped value if a match was found, otherwise self.

    Examples:
        >>> # Single mapping to Enum
        >>> Err(ValueError("fail")).map_exc({ValueError: ErrorCode.INVALID})
        Err(<ErrorCode.INVALID: 'invalid'>)

        >>> # Multiple mappings
        >>> mapping = {ValueError: ErrorCode.INVALID, KeyError: ErrorCode.MISSING}
        >>> Err(KeyError("key")).map_exc(mapping)
        Err(<ErrorCode.MISSING: 'missing'>)

    """
    err_type = type(self._error)
    if err_type in mapping:
        return Err(mapping[cast("Any", err_type)])
    return self

map_or

map_or(default: U, _func: object) -> U

Ignore the mapping and return the default.

Parameters:

Name	Type	Description	Default
default	U	The fallback value.	required
_func	object	Ignored mapping function.	required

Returns:

Type	Description
U	The default value.

Examples:

>>> Err("fail").map_or(0, lambda x: x * 2)
0

Source code in src/result/result.py

def map_or[U](self, default: U, _func: object) -> U:
    """Ignore the mapping and return the default.

    Args:
        default: The fallback value.
        _func: Ignored mapping function.

    Returns:
        The default value.

    Examples:
        >>> Err("fail").map_or(0, lambda x: x * 2)
        0

    """
    return default

map_or_else

map_or_else(
    default_func: Callable[[], U], _func: object
) -> U

Ignore the mapping and compute the default.

Parameters:

Name	Type	Description	Default
default_func	Callable[[], U]	Function to generate the fallback value.	required
_func	object	Ignored mapping function.	required

Returns:

Type	Description
U	The computed default value.

Examples:

>>> Err("fail").map_or_else(lambda: 0, lambda x: x * 2)
0

Source code in src/result/result.py

def map_or_else[U](self, default_func: Callable[[], U], _func: object) -> U:
    """Ignore the mapping and compute the default.

    Args:
        default_func: Function to generate the fallback value.
        _func: Ignored mapping function.

    Returns:
        The computed default value.

    Examples:
        >>> Err("fail").map_or_else(lambda: 0, lambda x: x * 2)
        0

    """
    return default_func()

match

match(
    on_ok: Callable[[T_local], U],
    on_err: Callable[[E_co], U],
) -> U

Exhaustively handle both success and failure cases.

Parameters:

Name	Type	Description	Default
on_ok	Callable[[T_local], U]	Ignored success handler.	required
on_err	Callable[[E_co], U]	Function to call with the error state.	required

Returns:

Type	Description
U	The result of the error handler.

Examples:

>>> res = Err("fail")
>>> res.match(on_ok=lambda x: x * 2, on_err=lambda e: 0)
0

Source code in src/result/result.py

def match[U, T_local](self, on_ok: Callable[[T_local], U], on_err: Callable[[E_co], U]) -> U:  # noqa: ARG002
    """Exhaustively handle both success and failure cases.

    Args:
        on_ok: Ignored success handler.
        on_err: Function to call with the error state.

    Returns:
        The result of the error handler.

    Examples:
        >>> res = Err("fail")
        >>> res.match(on_ok=lambda x: x * 2, on_err=lambda e: 0)
        0

    """
    return on_err(self._error)

ok

ok() -> object | None

Convert to Optional[T]. Always returns None for Err variants.

Examples:

>>> Err("fail").ok()
None

Source code in src/result/result.py

def ok(self) -> object | None:
    """Convert to Optional[T]. Always returns `None` for `Err` variants.

    Examples:
        >>> Err("fail").ok()
        None

    """
    return None

or_else

or_else(
    func: Callable[[E_co], Result[T_local, F_local]],
) -> Result[T_local, F_local]

Call a function with the error to attempt recovery.

Parameters:

Name	Type	Description	Default
func	Callable[[E_co], Result[T_local, F_local]]	A function that takes the error and returns a new Result.	required

Returns:

Type	Description
Result[T_local, F_local]	The Result returned by func.

Examples:

>>> Err("file missing").or_else(lambda _: Ok("default content"))
Ok('default content')

Source code in src/result/result.py

def or_else[T_local, F_local](self, func: Callable[[E_co], Result[T_local, F_local]]) -> Result[T_local, F_local]:
    """Call a function with the error to attempt recovery.

    Args:
        func: A function that takes the error and returns a new `Result`.

    Returns:
        The `Result` returned by `func`.

    Examples:
        >>> Err("file missing").or_else(lambda _: Ok("default content"))
        Ok('default content')

    """
    return func(self._error)

product

product(
    _other: Result[U, E2],
) -> Result[tuple[Any, U], E_co | E2]

Short-circuit the product and return self.

Parameters:

Name	Type	Description	Default
_other	Result[U, E2]	Ignored result.	required

Returns:

Type	Description
Result[tuple[Any, U], E_co | E2]	Self unchanged.

Source code in src/result/result.py

def product[U, E2](self, _other: Result[U, E2]) -> Result[tuple[Any, U], E_co | E2]:
    """Short-circuit the product and return self.

    Args:
        _other: Ignored result.

    Returns:
        Self unchanged.

    """
    return cast("Any", self)

replace

replace(_value: object) -> Err[E_co]

Ignore the replacement and return self unchanged.

Source code in src/result/result.py

def replace(self, _value: object) -> Err[E_co]:
    """Ignore the replacement and return self unchanged."""
    return self

replace_err

replace_err(error: F) -> Err[F]

Discard the contained error and replace it with a constant.

Parameters:

Name	Type	Description	Default
error	F	The new error to wrap in Err.	required

Returns:

Type	Description
Err[F]	A new Err instance.

Examples:

>>> Err("timeout").replace_err("network error")
Err('network error')

Source code in src/result/result.py

def replace_err[F](self, error: F) -> Err[F]:
    """Discard the contained error and replace it with a constant.

    Args:
        error: The new error to wrap in Err.

    Returns:
        A new `Err` instance.

    Examples:
        >>> Err("timeout").replace_err("network error")
        Err('network error')

    """
    return Err(error)

tap

tap(_func: Callable[[Any], Any]) -> Err[E_co]

Ignore the side effect and return self unchanged.

Source code in src/result/result.py

def tap(self, _func: Callable[[Any], Any]) -> Err[E_co]:
    """Ignore the side effect and return self unchanged."""
    return self

tap_async async

tap_async(
    _func: Callable[[T_local], Awaitable[Any]],
) -> Err[E_co]

Ignore async success side effects.

Parameters:

Name	Type	Description	Default
_func	Callable[[T_local], Awaitable[Any]]	Ignored async side effect function.	required

Returns:

Type	Description
Err[E_co]	The current instance unchanged.

Source code in src/result/result.py

async def tap_async[T_local](self, _func: Callable[[T_local], Awaitable[Any]]) -> Err[E_co]:
    """Ignore async success side effects.

    Args:
        _func: Ignored async side effect function.

    Returns:
        The current instance unchanged.

    """
    return self

tap_err

tap_err(func: Callable[[E_co], Any]) -> Err[E_co]

Call a function with the contained error for side effects.

Parameters:

Name	Type	Description	Default
func	Callable[[E_co], Any]	A function called with the error state.	required

Returns:

Type	Description
Err[E_co]	The current instance unchanged.

Examples:

>>> Err("db fail").tap_err(print)
db fail
Err('db fail')

Source code in src/result/result.py

def tap_err(self, func: Callable[[E_co], Any]) -> Err[E_co]:
    """Call a function with the contained error for side effects.

    Args:
        func: A function called with the error state.

    Returns:
        The current instance unchanged.

    Examples:
        >>> Err("db fail").tap_err(print)
        db fail
        Err('db fail')

    """
    func(self._error)
    return self

transpose

transpose() -> Err[E_co]

Transpose an Err variant.

Err(e) remains Err(e) (wrapped in the optional result).

Returns:

Type	Description
Err[E_co]	Self unchanged.

Source code in src/result/result.py

def transpose(self) -> Err[E_co]:
    """Transpose an Err variant.

    Err(e) remains Err(e) (wrapped in the optional result).

    Returns:
        Self unchanged.

    """
    return self

unwrap

unwrap() -> Never

Root-level unwrap is disabled. Use .unsafe.unwrap() instead.

Source code in src/result/result.py

def unwrap(self) -> Never:
    """Root-level unwrap is disabled. Use .unsafe.unwrap() instead."""
    _raise_api_error("unwrap")

unwrap_err

unwrap_err() -> Never

Root-level unwrap_err is disabled. Use .unsafe.unwrap_err() instead.

Source code in src/result/result.py

def unwrap_err(self) -> Never:
    """Root-level unwrap_err is disabled. Use .unsafe.unwrap_err() instead."""
    _raise_api_error("unwrap_err")

unwrap_or

unwrap_or(default: T_local) -> T_local

Return the provided default value.

Parameters:

Name	Type	Description	Default
default	T_local	The fallback value.	required

Returns:

Type	Description
T_local	The default value.

Examples:

>>> Err("fail").unwrap_or(42)
42

Source code in src/result/result.py

def unwrap_or[T_local](self, default: T_local) -> T_local:
    """Return the provided default value.

    Args:
        default: The fallback value.

    Returns:
        The `default` value.

    Examples:
        >>> Err("fail").unwrap_or(42)
        42

    """
    return default

unwrap_or_else

unwrap_or_else(func: Callable[[E_co], T_local]) -> T_local

Call a function with the error to produce a fallback value.

Parameters:

Name	Type	Description	Default
func	Callable[[E_co], T_local]	A function that generates a fallback value from the error.	required

Returns:

Type	Description
T_local	The result of func(error).

Examples:

>>> Err("fail").unwrap_or_else(lambda e: f"recovered from {e}")
'recovered from fail'

Source code in src/result/result.py

def unwrap_or_else[T_local](self, func: Callable[[E_co], T_local]) -> T_local:
    """Call a function with the error to produce a fallback value.

    Args:
        func: A function that generates a fallback value from the error.

    Returns:
        The result of `func(error)`.

    Examples:
        >>> Err("fail").unwrap_or_else(lambda e: f"recovered from {e}")
        'recovered from fail'

    """
    return func(self._error)

unwrap_or_raise

unwrap_or_raise(_e: type[Exception]) -> Never

Root-level unwrap_or_raise is disabled. Use .unsafe.unwrap_or_raise() instead.

Source code in src/result/result.py

def unwrap_or_raise(self, _e: type[Exception]) -> Never:
    """Root-level unwrap_or_raise is disabled. Use .unsafe.unwrap_or_raise() instead."""
    _raise_api_error("unwrap_or_raise")

Ok dataclass

Ok(value: T_co)

A container representing a successful computation result.

To access the value safely, use pattern matching, functional chaining, or the .ok() conversion method.

Examples:

>>> res = Ok(200)
>>> match res:
...     case Ok(v):
...         print(f"Success: {v}")
Success: 200

Initialize an Ok variant.

Parameters:

Name	Type	Description	Default
value	T_co	The successful result to wrap.	required

Source code in src/result/result.py

def __init__(self, value: T_co) -> None:
    """Initialize an Ok variant.

    Args:
        value: The successful result to wrap.

    """
    object.__setattr__(self, "_value", value)

unsafe property

unsafe: _OkUnsafe[T_co]

Namespace for operations that might panic (raise exceptions).

Examples:

>>> Ok(10).unsafe.unwrap()
10

value property

value: T_co

Access the success value.

Note: While accessible on the Ok variant, it is recommended to use pattern matching or functional methods for maximum safety.

__add__

__add__(other: Any) -> Result[Any, Any]

Support short-circuiting addition (+) operator.

If both are Ok, returns Ok(a + b). If other is Err, returns other.

Source code in src/result/result.py

def __add__(self, other: Any) -> Result[Any, Any]:
    """Support short-circuiting addition (+) operator.

    If both are Ok, returns Ok(a + b). If other is Err, returns other.
    """
    if isinstance(other, Ok):
        return Ok(self._value + other._value)  # pyright: ignore[reportUnknownArgumentType, reportUnknownMemberType]
    if isinstance(other, Err):
        return other  # pyright: ignore[reportUnknownVariableType]
    return NotImplemented

__aiter__ async

__aiter__() -> AsyncGenerator[T_co, Any]

Allow use in async generator expressions for do-notation.

Source code in src/result/result.py

async def __aiter__(self) -> AsyncGenerator[T_co, Any]:
    """Allow use in async generator expressions for do-notation."""
    yield self._value

__bool__

__bool__() -> Literal[True]

Truthiness check. Returns True for Ok.

This enables idiomatic Python conditional narrowing: if res: # type is narrowed to Ok

Source code in src/result/result.py

def __bool__(self) -> Literal[True]:
    """Truthiness check. Returns True for Ok.

    This enables idiomatic Python conditional narrowing:
    `if res: # type is narrowed to Ok`
    """
    return True

__getattr__

__getattr__(name: str) -> Never

Educational runtime safeguard against common API mistakes.

Source code in src/result/result.py

def __getattr__(self, name: str) -> Never:
    """Educational runtime safeguard against common API mistakes."""
    if name in {
        "value",
        "error",
        "unwrap",
        "unwrap_err",
        "expect",
        "expect_err",
        "unwrap_or_raise",
        "inspect",
        "inspect_async",
        "inspect_err",
    }:
        _raise_api_error(name)
    msg = f"'{self.__class__.__name__}' object has no attribute '{name}'"
    raise AttributeError(msg)

__hash__

__hash__() -> int

Explicit hash to prevent collisions with Err variants.

Source code in src/result/result.py

def __hash__(self) -> int:
    """Explicit hash to prevent collisions with Err variants."""
    return hash((Ok, self._value))

__iter__

__iter__() -> Generator[Any, Any, T_co]

Allow use in generator expressions for do-notation.

Source code in src/result/result.py

def __iter__(self) -> Generator[Any, Any, T_co]:
    """Allow use in generator expressions for do-notation."""
    yield self._value
    return self._value  # noqa: B901

and_then

and_then(
    func: Callable[[T_co], Result[U, E_local]],
) -> Result[U, E_local]

Chain another result-returning operation (FlatMap).

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], Result[U, E_local]]	A function that takes the value and returns a new Result.	required

Returns:

Type	Description
Result[U, E_local]	The Result returned by func.

Examples:

>>> def validate(n: int) -> Result[int, str]:
...     return Ok(n) if n > 0 else Err("too small")
>>> Ok(10).and_then(validate)
Ok(10)

Source code in src/result/result.py

def and_then[U, E_local](self, func: Callable[[T_co], Result[U, E_local]]) -> Result[U, E_local]:
    """Chain another result-returning operation (FlatMap).

    Args:
        func: A function that takes the value and returns a new `Result`.

    Returns:
        The `Result` returned by `func`.

    Examples:
        >>> def validate(n: int) -> Result[int, str]:
        ...     return Ok(n) if n > 0 else Err("too small")
        >>> Ok(10).and_then(validate)
        Ok(10)

    """
    return func(self._value)

and_then_async async

and_then_async(
    func: Callable[[T_co], Awaitable[Result[U, E_local]]],
) -> Result[U, E_local]

Chain an async computation that might fail.

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], Awaitable[Result[U, E_local]]]	An async function that takes the value and returns a new Result.	required

Returns:

Type	Description
Result[U, E_local]	The Result returned by func.

Examples:

>>> async def check(n):
...     return Ok(n) if n > 0 else Err("low")
>>> await Ok(10).and_then_async(check)
Ok(10)

Source code in src/result/result.py

async def and_then_async[U, E_local](
    self, func: Callable[[T_co], Awaitable[Result[U, E_local]]]
) -> Result[U, E_local]:
    """Chain an async computation that might fail.

    Args:
        func: An async function that takes the value and returns a new Result.

    Returns:
        The Result returned by func.

    Examples:
        >>> async def check(n):
        ...     return Ok(n) if n > 0 else Err("low")
        >>> await Ok(10).and_then_async(check)
        Ok(10)

    """
    return await func(self._value)

cast_types

cast_types() -> Result[U, F]

Zero-runtime-cost type hint override for strict variance edge cases.

This allows manually guiding the type checker when it fails to infer complex union types correctly.

Returns:

Type	Description
Result[U, F]	The same instance, but with new type parameters for the checker.

Source code in src/result/result.py

def cast_types[U, F](self) -> Result[U, F]:
    """Zero-runtime-cost type hint override for strict variance edge cases.

    This allows manually guiding the type checker when it fails to infer
    complex union types correctly.

    Returns:
        The same instance, but with new type parameters for the checker.

    """
    return cast("Result[U, F]", self)

err

err() -> object | None

Convert to Optional[E].

Returns:

Type	Description
object | None	Always returns None for Ok variants.

Examples:

>>> Ok(10).err()
None

Source code in src/result/result.py

def err(self) -> object | None:
    """Convert to Optional[E].

    Returns:
        Always returns `None` for `Ok` variants.

    Examples:
        >>> Ok(10).err()
        None

    """
    return None

expect

expect(_msg: str) -> Never

Root-level expect is disabled. Use .unsafe.expect() instead.

Source code in src/result/result.py

def expect(self, _msg: str) -> Never:
    """Root-level expect is disabled. Use .unsafe.expect() instead."""
    _raise_api_error("expect")

expect_err

expect_err(_msg: str) -> Never

Root-level expect_err is disabled. Use .unsafe.expect_err() instead.

Source code in src/result/result.py

def expect_err(self, _msg: str) -> Never:
    """Root-level expect_err is disabled. Use .unsafe.expect_err() instead."""
    _raise_api_error("expect_err")

filter

filter(
    predicate: Callable[[T_co], bool], error: E_local
) -> Result[T_co, E_local]

Convert success to failure if a condition is not met.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T_co], bool]	A function that returns True for valid values.	required
error	E_local	The error state to use if the predicate returns False.	required

Returns:

Type	Description
Result[T_co, E_local]	Self if predicate is True, otherwise Err(error).

Examples:

>>> Ok(10).filter(lambda x: x > 5, "error")
Ok(10)
>>> Ok(3).filter(lambda x: x > 5, "too small")
Err('too small')

Source code in src/result/result.py

def filter[E_local](self, predicate: Callable[[T_co], bool], error: E_local) -> Result[T_co, E_local]:
    """Convert success to failure if a condition is not met.

    Args:
        predicate: A function that returns True for valid values.
        error: The error state to use if the predicate returns False.

    Returns:
        Self if predicate is True, otherwise `Err(error)`.

    Examples:
        >>> Ok(10).filter(lambda x: x > 5, "error")
        Ok(10)
        >>> Ok(3).filter(lambda x: x > 5, "too small")
        Err('too small')

    """
    if predicate(self._value):
        return self
    return Err(error)

flatten

flatten() -> R

flatten() -> Ok[T_co]

flatten() -> Result[Any, Any]

Flatten a nested Result.

If the contained value is itself a Result (Ok or Err), it is returned. Otherwise, the current instance is returned unchanged.

Returns:

Type	Description
Result[Any, Any]	The inner Result or self.

Examples:

>>> Ok(Ok(10)).flatten()
Ok(10)
>>> Ok(Err("fail")).flatten()
Err('fail')

Source code in src/result/result.py

def flatten(self) -> Result[Any, Any]:
    """Flatten a nested Result.

    If the contained value is itself a `Result` (Ok or Err), it is returned.
    Otherwise, the current instance is returned unchanged.

    Returns:
        The inner Result or self.

    Examples:
        >>> Ok(Ok(10)).flatten()
        Ok(10)
        >>> Ok(Err("fail")).flatten()
        Err('fail')

    """
    val = self._value
    if isinstance(val, Ok | Err):
        return cast("Result[Any, Any]", val)
    return self

inspect

inspect(_func: Callable[[Any], Any]) -> Never

Root-level inspect is disabled. Use .tap() instead.

Source code in src/result/result.py

def inspect(self, _func: Callable[[Any], Any]) -> Never:
    """Root-level inspect is disabled. Use .tap() instead."""
    _raise_api_error("inspect")

inspect_async

inspect_async(
    _func: Callable[[Any], Awaitable[Any]],
) -> Never

Root-level inspect_async is disabled. Use .tap_async() instead.

Source code in src/result/result.py

def inspect_async(self, _func: Callable[[Any], Awaitable[Any]]) -> Never:
    """Root-level inspect_async is disabled. Use .tap_async() instead."""
    _raise_api_error("inspect_async")

inspect_err

inspect_err(_func: Callable[[Any], Any]) -> Never

Root-level inspect_err is disabled. Use .tap_err() instead.

Source code in src/result/result.py

def inspect_err(self, _func: Callable[[Any], Any]) -> Never:
    """Root-level inspect_err is disabled. Use .tap_err() instead."""
    _raise_api_error("inspect_err")

is_err

is_err() -> Literal[False]

Check if the result is an Err variant.

Returns:

Type	Description
Literal[False]	Always False for Ok instances.

Source code in src/result/result.py

def is_err(self) -> Literal[False]:
    """Check if the result is an Err variant.

    Returns:
        Always False for Ok instances.

    """
    return False

is_err_and

is_err_and(
    _predicate: Callable[[Any], bool],
) -> Literal[False]

Check if the result is an Err variant and matches a predicate.

Always returns False for Ok instances.

Parameters:

Name	Type	Description	Default
_predicate	Callable[[Any], bool]	Ignored for Ok instances.	required

Returns:

Type	Description
Literal[False]	Always False.

Source code in src/result/result.py

def is_err_and(self, _predicate: Callable[[Any], bool]) -> Literal[False]:
    """Check if the result is an Err variant and matches a predicate.

    Always returns False for Ok instances.

    Args:
        _predicate: Ignored for Ok instances.

    Returns:
        Always False.

    """
    return False

is_ok

is_ok() -> Literal[True]

Check if the result is an Ok variant.

Returns:

Type	Description
Literal[True]	Always True for Ok instances.

Source code in src/result/result.py

def is_ok(self) -> Literal[True]:
    """Check if the result is an Ok variant.

    Returns:
        Always True for Ok instances.

    """
    return True

is_ok_and

is_ok_and(predicate: Callable[[T_co], bool]) -> bool

Check if the result is an Ok variant and matches a predicate.

Parameters:

Name	Type	Description	Default
predicate	Callable[[T_co], bool]	A function to test the success value.	required

Returns:

Type	Description
bool	True if the result is Ok and the predicate returns True.

Examples:

>>> Ok(10).is_ok_and(lambda x: x > 5)
True

Source code in src/result/result.py

def is_ok_and(self, predicate: Callable[[T_co], bool]) -> bool:
    """Check if the result is an Ok variant and matches a predicate.

    Args:
        predicate: A function to test the success value.

    Returns:
        True if the result is Ok and the predicate returns True.

    Examples:
        >>> Ok(10).is_ok_and(lambda x: x > 5)
        True

    """
    return predicate(self._value)

map

map(func: Callable[[T_co], U]) -> Ok[U]

Apply a function to the contained value.

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], U]	A pure function to transform the success value.	required

Returns:

Type	Description
Ok[U]	A new Ok containing the transformed value.

Examples:

>>> Ok(10).map(lambda x: x * 2)
Ok(20)

Source code in src/result/result.py

def map[U](self, func: Callable[[T_co], U]) -> Ok[U]:
    """Apply a function to the contained value.

    Args:
        func: A pure function to transform the success value.

    Returns:
        A new `Ok` containing the transformed value.

    Examples:
        >>> Ok(10).map(lambda x: x * 2)
        Ok(20)

    """
    return Ok(func(self._value))

map_async async

map_async(func: Callable[[T_co], Awaitable[U]]) -> Ok[U]

Apply an async function to the contained value.

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], Awaitable[U]]	An async function to transform the success value.	required

Returns:

Type	Description
Ok[U]	A new Ok containing the transformed value.

Examples:

>>> async def double(x):
...     return x * 2
>>> await Ok(10).map_async(double)
Ok(20)

Source code in src/result/result.py

async def map_async[U](self, func: Callable[[T_co], Awaitable[U]]) -> Ok[U]:
    """Apply an async function to the contained value.

    Args:
        func: An async function to transform the success value.

    Returns:
        A new Ok containing the transformed value.

    Examples:
        >>> async def double(x):
        ...     return x * 2
        >>> await Ok(10).map_async(double)
        Ok(20)

    """
    return Ok(await func(self._value))

map_err

map_err(_func: Callable[[Any], Any]) -> Ok[T_co]

Ignore the error mapping and return self unchanged.

Source code in src/result/result.py

def map_err(self, _func: Callable[[Any], Any]) -> Ok[T_co]:
    """Ignore the error mapping and return self unchanged."""
    return self

map_exc

map_exc(
    _mapping: Mapping[type[Exception], Any],
) -> Ok[T_co]

Ignore the error mapping and return self unchanged.

Parameters:

Name	Type	Description	Default
_mapping	Mapping[type[Exception], Any]	A dictionary mapping exception types to new values.	required

Returns:

Type	Description
Ok[T_co]	The current instance unchanged.

Examples:

>>> Ok(10).map_exc({ValueError: ErrorCode.INVALID})
Ok(10)

Source code in src/result/result.py

def map_exc(self, _mapping: Mapping[type[Exception], Any]) -> Ok[T_co]:
    """Ignore the error mapping and return self unchanged.

    Args:
        _mapping: A dictionary mapping exception types to new values.

    Returns:
        The current instance unchanged.

    Examples:
        >>> Ok(10).map_exc({ValueError: ErrorCode.INVALID})
        Ok(10)

    """
    return self

map_or

map_or(_default: U, func: Callable[[T_co], U]) -> U

Apply a function to the value or return a default.

Parameters:

Name	Type	Description	Default
_default	U	The fallback value used if this were an Err.	required
func	Callable[[T_co], U]	A function to transform the success value.	required

Returns:

Type	Description
U	The transformed value.

Examples:

>>> Ok(10).map_or(0, lambda x: x * 2)
20

Source code in src/result/result.py

def map_or[U](self, _default: U, func: Callable[[T_co], U]) -> U:
    """Apply a function to the value or return a default.

    Args:
        _default: The fallback value used if this were an Err.
        func: A function to transform the success value.

    Returns:
        The transformed value.

    Examples:
        >>> Ok(10).map_or(0, lambda x: x * 2)
        20

    """
    return func(self._value)

map_or_else

map_or_else(
    _default_func: Callable[[], U],
    func: Callable[[T_co], U],
) -> U

Apply a function to the value or compute a default.

Parameters:

Name	Type	Description	Default
_default_func	Callable[[], U]	A function to generate a fallback value.	required
func	Callable[[T_co], U]	A function to transform the success value.	required

Returns:

Type	Description
U	The transformed value.

Examples:

>>> Ok(10).map_or_else(lambda: 0, lambda x: x * 2)
20

Source code in src/result/result.py

def map_or_else[U](self, _default_func: Callable[[], U], func: Callable[[T_co], U]) -> U:
    """Apply a function to the value or compute a default.

    Args:
        _default_func: A function to generate a fallback value.
        func: A function to transform the success value.

    Returns:
        The transformed value.

    Examples:
        >>> Ok(10).map_or_else(lambda: 0, lambda x: x * 2)
        20

    """
    return func(self._value)

match

match(
    on_ok: Callable[[T_co], U], on_err: Callable[[Any], U]
) -> U

Exhaustively handle both success and failure cases.

Parameters:

Name	Type	Description	Default
on_ok	Callable[[T_co], U]	Function to call with the success value.	required
on_err	Callable[[Any], U]	Function to call with the error state.	required

Returns:

Type	Description
U	The result of whichever function was called.

Examples:

>>> res = Ok(10)
>>> res.match(on_ok=lambda x: x * 2, on_err=lambda e: 0)
20

Source code in src/result/result.py

def match[U](self, on_ok: Callable[[T_co], U], on_err: Callable[[Any], U]) -> U:  # noqa: ARG002
    """Exhaustively handle both success and failure cases.

    Args:
        on_ok: Function to call with the success value.
        on_err: Function to call with the error state.

    Returns:
        The result of whichever function was called.

    Examples:
        >>> res = Ok(10)
        >>> res.match(on_ok=lambda x: x * 2, on_err=lambda e: 0)
        20

    """
    return on_ok(self._value)

ok

ok() -> T_co

Convert to Optional[T].

Returns:

Type	Description
T_co	The contained value. Always succeeds on Ok.

Examples:

>>> Ok(10).ok()
10

Source code in src/result/result.py

def ok(self) -> T_co:
    """Convert to Optional[T].

    Returns:
        The contained value. Always succeeds on `Ok`.

    Examples:
        >>> Ok(10).ok()
        10

    """
    return self._value

or_else

or_else(_func: object) -> Ok[T_co]

Ignore the recovery function and return self unchanged.

Parameters:

Name	Type	Description	Default
_func	object	Ignored recovery function.	required

Returns:

Type	Description
Ok[T_co]	The current instance unchanged.

Examples:

>>> Ok(10).or_else(lambda e: Ok(0))
Ok(10)

Source code in src/result/result.py

def or_else(self, _func: object) -> Ok[T_co]:
    """Ignore the recovery function and return self unchanged.

    Args:
        _func: Ignored recovery function.

    Returns:
        The current instance unchanged.

    Examples:
        >>> Ok(10).or_else(lambda e: Ok(0))
        Ok(10)

    """
    return self

product

product(
    other: Result[U, E2],
) -> Result[tuple[T_co, U], E_co | E2]

Combine two results into a result of a tuple (Zip).

Parameters:

Name	Type	Description	Default
other	Result[U, E2]	Another Result to zipping with.	required

Returns:

Type	Description
Result[tuple[T_co, U], E_co | E2]	Ok((val1, val2)) if both are Ok, otherwise the first Err.

Examples:

>>> Ok(1).product(Ok(2))
Ok((1, 2))

Source code in src/result/result.py

def product[U, E2](self, other: Result[U, E2]) -> Result[tuple[T_co, U], E_co | E2]:  # type: ignore[valid-type]  # ty:ignore[unused-type-ignore-comment, unused-ignore-comment]
    """Combine two results into a result of a tuple (Zip).

    Args:
        other: Another Result to zipping with.

    Returns:
        Ok((val1, val2)) if both are Ok, otherwise the first Err.

    Examples:
        >>> Ok(1).product(Ok(2))
        Ok((1, 2))

    """
    if isinstance(other, Err):
        return cast("Any", other)
    return Ok[tuple[T_co, U]]((self._value, other._value))

replace

replace(value: U) -> Ok[U]

Discard the contained value and replace it with a constant.

Parameters:

Name	Type	Description	Default
value	U	The new value to wrap in Ok.	required

Examples:

>>> Ok(10).replace("done")
Ok('done')

Source code in src/result/result.py

def replace[U](self, value: U) -> Ok[U]:
    """Discard the contained value and replace it with a constant.

    Args:
        value: The new value to wrap in Ok.

    Examples:
        >>> Ok(10).replace("done")
        Ok('done')

    """
    return Ok(value)

replace_err

replace_err(_error: object) -> Ok[T_co]

Ignore the error replacement and return self unchanged.

Source code in src/result/result.py

def replace_err(self, _error: object) -> Ok[T_co]:
    """Ignore the error replacement and return self unchanged."""
    return self

tap

tap(func: Callable[[T_co], Any]) -> Ok[T_co]

Call a function with the contained value for side effects.

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], Any]	A function called with the success value.	required

Returns:

Type	Description
Ok[T_co]	The current instance unchanged.

Examples:

>>> Ok(10).tap(print).map(lambda x: x + 1)
10
Ok(11)

Source code in src/result/result.py

def tap(self, func: Callable[[T_co], Any]) -> Ok[T_co]:
    """Call a function with the contained value for side effects.

    Args:
        func: A function called with the success value.

    Returns:
        The current instance unchanged.

    Examples:
        >>> Ok(10).tap(print).map(lambda x: x + 1)
        10
        Ok(11)

    """
    func(self._value)
    return self

tap_async async

tap_async(
    func: Callable[[T_co], Awaitable[Any]],
) -> Ok[T_co]

Call an async function with the contained value for side effects.

Parameters:

Name	Type	Description	Default
func	Callable[[T_co], Awaitable[Any]]	An async function called with the success value.	required

Returns:

Type	Description
Ok[T_co]	The current instance unchanged.

Examples:

>>> async def log(x):
...     print(f"Logging: {x}")
>>> await Ok(10).tap_async(log)
Logging: 10
Ok(10)

Source code in src/result/result.py

async def tap_async(self, func: Callable[[T_co], Awaitable[Any]]) -> Ok[T_co]:
    """Call an async function with the contained value for side effects.

    Args:
        func: An async function called with the success value.

    Returns:
        The current instance unchanged.

    Examples:
        >>> async def log(x):
        ...     print(f"Logging: {x}")
        >>> await Ok(10).tap_async(log)
        Logging: 10
        Ok(10)

    """
    await func(self._value)
    return self

tap_err

tap_err(_func: Callable[[Any], Any]) -> Ok[T_co]

Ignore the error side effect and return self unchanged.

Source code in src/result/result.py

def tap_err(self, _func: Callable[[Any], Any]) -> Ok[T_co]:
    """Ignore the error side effect and return self unchanged."""
    return self

transpose

transpose() -> Ok[U] | None

transpose() -> Ok[T_co] | None

transpose() -> Ok[Any] | None

Transpose a Result of an Optional into an Optional of a Result.

Ok(None) becomes None. Ok(Some(v)) becomes Ok(v).

Returns:

Type	Description
Ok[Any] | None	None if the value is None, otherwise self.

Examples:

>>> Ok(10).transpose()
Ok(10)
>>> Ok(None).transpose()
None

Source code in src/result/result.py

def transpose(self) -> Ok[Any] | None:
    """Transpose a Result of an Optional into an Optional of a Result.

    Ok(None) becomes None. Ok(Some(v)) becomes Ok(v).

    Returns:
        None if the value is None, otherwise self.

    Examples:
        >>> Ok(10).transpose()
        Ok(10)
        >>> Ok(None).transpose()
        None

    """
    if self._value is None:
        return None
    return cast("Ok[Any]", self)

unwrap

unwrap() -> Never

Root-level unwrap is disabled. Use .unsafe.unwrap() instead.

Source code in src/result/result.py

def unwrap(self) -> Never:
    """Root-level unwrap is disabled. Use .unsafe.unwrap() instead."""
    _raise_api_error("unwrap")

unwrap_err

unwrap_err() -> Never

Root-level unwrap_err is disabled. Use .unsafe.unwrap_err() instead.

Source code in src/result/result.py

def unwrap_err(self) -> Never:
    """Root-level unwrap_err is disabled. Use .unsafe.unwrap_err() instead."""
    _raise_api_error("unwrap_err")

unwrap_or

unwrap_or(_default: object) -> T_co

Extract the contained value, ignoring the default.

Parameters:

Name	Type	Description	Default
_default	object	Ignored fallback value.	required

Returns:

Type	Description
T_co	The contained success value.

Examples:

>>> Ok(10).unwrap_or(0)
10

Source code in src/result/result.py

def unwrap_or(self, _default: object) -> T_co:
    """Extract the contained value, ignoring the default.

    Args:
        _default: Ignored fallback value.

    Returns:
        The contained success value.

    Examples:
        >>> Ok(10).unwrap_or(0)
        10

    """
    return self._value

unwrap_or_else

unwrap_or_else(_func: Callable[[Any], T_co]) -> T_co

Extract the contained value, ignoring the fallback function.

Parameters:

Name	Type	Description	Default
_func	Callable[[Any], T_co]	Ignored fallback function.	required

Returns:

Type	Description
T_co	The contained success value.

Examples:

>>> Ok(10).unwrap_or_else(lambda e: 0)
10

Source code in src/result/result.py

def unwrap_or_else(self, _func: Callable[[Any], T_co]) -> T_co:
    """Extract the contained value, ignoring the fallback function.

    Args:
        _func: Ignored fallback function.

    Returns:
        The contained success value.

    Examples:
        >>> Ok(10).unwrap_or_else(lambda e: 0)
        10

    """
    return self._value

unwrap_or_raise

unwrap_or_raise(_e: type[Exception]) -> Never

Root-level unwrap_or_raise is disabled. Use .unsafe.unwrap_or_raise() instead.

Source code in src/result/result.py

def unwrap_or_raise(self, _e: type[Exception]) -> Never:
    """Root-level unwrap_or_raise is disabled. Use .unsafe.unwrap_or_raise() instead."""
    _raise_api_error("unwrap_or_raise")

UnwrapError

UnwrapError(result: Result[Any, Any], message: str)

Bases: RuntimeError

Exception raised when an unsafe unwrap operation fails.

Attributes:

Name	Type	Description
result		The Result instance that failed to unwrap.

Source code in src/result/result.py

def __init__(self, result: Result[Any, Any], message: str) -> None:
    super().__init__(message)
    self.result = result

any_ok

any_ok(
    results: Iterable[Result[T_local, E_local]],
) -> Result[T_local, list[E_local]]

Return the first Ok variant found, or a list of all errors if none succeeded.

Parameters:

Name	Type	Description	Default
results	Iterable[Result[T_local, E_local]]	An iterable of Results.	required

Returns:

Type	Description
Result[T_local, list[E_local]]	The first Ok found, or Err containing a list of all encountered errors.

Examples:

>>> any_ok([Err("a"), Ok(1), Ok(2)])
Ok(1)
>>> any_ok([Err("a"), Err("b")])
Err(['a', 'b'])

Source code in src/result/result.py

def any_ok[T_local, E_local](results: Iterable[Result[T_local, E_local]]) -> Result[T_local, list[E_local]]:
    """Return the first Ok variant found, or a list of all errors if none succeeded.

    Args:
        results: An iterable of Results.

    Returns:
        The first Ok found, or Err containing a list of all encountered errors.

    Examples:
        >>> any_ok([Err("a"), Ok(1), Ok(2)])
        Ok(1)
        >>> any_ok([Err("a"), Err("b")])
        Err(['a', 'b'])

    """
    errs: list[E_local] = []
    for res in results:
        if isinstance(res, Ok):
            return res
        errs.append(res._error)
    return Err(errs)

as_err

as_err(
    exception: E_in,
    mapping: type[Exception]
    | tuple[type[Exception], ...]
    | Mapping[type[Exception], E_out]
    | None = None,
    *,
    map_to: E_out | None = None,
) -> Err[E_in | E_out]

Lift a caught exception into an Err variant with optional mapping.

This pinpoint utility is ideal for manual conversion inside standard try/except blocks. It uses the same mapping logic as the @catch decorator.

Parameters:

Name	Type	Description	Default
exception	E_in	The exception instance to wrap.	required
mapping	type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], E_out] | None	Optional exception type, tuple, or dict for transformation.	None
map_to	E_out | None	Optional value to use as the error if mapping is a type/tuple.	None

Returns:

Type	Description
Err[E_in | E_out]	An Err variant containing the (potentially mapped) error.

Examples:

>>> try:
...     raise ValueError("fail")
... except Exception as e:
...     res = as_err(e, {ValueError: "invalid"})
>>> res
Err('invalid')

>>> # Using map_to
>>> try:
...     raise ValueError("fail")
... except Exception as e:
...     res = as_err(e, ValueError, map_to="mapped")
>>> res
Err('mapped')

Source code in src/result/result.py

def as_err[E_in: Exception, E_out](
    exception: E_in,
    mapping: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], E_out] | None = None,
    *,
    map_to: E_out | None = None,
) -> Err[E_in | E_out]:
    """Lift a caught exception into an Err variant with optional mapping.

    This pinpoint utility is ideal for manual conversion inside standard
    try/except blocks. It uses the same mapping logic as the @catch decorator.

    Args:
        exception: The exception instance to wrap.
        mapping: Optional exception type, tuple, or dict for transformation.
        map_to: Optional value to use as the error if mapping is a type/tuple.

    Returns:
        An Err variant containing the (potentially mapped) error.

    Examples:
        >>> try:
        ...     raise ValueError("fail")
        ... except Exception as e:
        ...     res = as_err(e, {ValueError: "invalid"})
        >>> res
        Err('invalid')

        >>> # Using map_to
        >>> try:
        ...     raise ValueError("fail")
        ... except Exception as e:
        ...     res = as_err(e, ValueError, map_to="mapped")
        >>> res
        Err('mapped')

    """
    if mapping is None:
        return Err(exception)

    exc_map = _resolve_mapping(mapping, map_to)
    has_mapping = map_to is not None or isinstance(mapping, Mapping)

    if type(exception) in exc_map:
        mapped = exc_map[type(exception)] if has_mapping else exception
        return Err(mapped)

    return Err(exception)

assert_ok

assert_ok(result_or_message: Result[T, E]) -> T

assert_ok(
    result_or_message: str = "Result was Err",
) -> AssertOk

assert_ok(result_or_message: Any = 'Result was Err') -> Any

A dual-purpose utility for asserting that a Result must be Ok.

Can be used as a standalone function or as a context manager. If a Result is an Err, it raises an AssertionError.

Functional Mode

When passed a Result, it returns the success value or raises AssertionError immediately. This is the high-performance way to assert invariants.

Context Manager Mode

When used as a context manager, it automatically monitors local variable assignments within the block using sys.settrace. If any local variable is assigned an Err variant, it raises an AssertionError (fail-fast).

Performance & Behavior Notes: - Overhead: The context manager installs a trace function, which introduces performance overhead compared to functional usage. Use it for scripts and prototypes rather than hot loops. - Scanning Scope: The automatic scanning only catches Err variants that are assigned to a variable name in the local scope. Unassigned return values will NOT be caught.

Examples:

>>> # 1. Functional usage (Low overhead)
>>> val = assert_ok(Ok(10))
>>> # assert_ok(Err("fail")) # Raises AssertionError

>>> # 2. Automatic context manager usage (Higher overhead, Fail-fast)
>>> with assert_ok("Critical operations"):
...     res = Ok(1)  # Fine
...     # res2 = Err("boom") # Raises AssertionError immediately

>>> # 3. Explicit check usage (Lower overhead in context)
>>> with assert_ok() as ctx:
...     val = ctx.check(Ok(42))

Source code in src/result/result.py

def assert_ok(result_or_message: Any = "Result was Err") -> Any:
    """A dual-purpose utility for asserting that a Result must be Ok.

    Can be used as a standalone function or as a context manager.
    If a Result is an Err, it raises an AssertionError.

    Functional Mode:
        When passed a `Result`, it returns the success value or raises
        `AssertionError` immediately. This is the high-performance way to
        assert invariants.

    Context Manager Mode:
        When used as a context manager, it automatically monitors local variable
        assignments within the block using `sys.settrace`. If any local variable
        is assigned an `Err` variant, it raises an `AssertionError` (fail-fast).

    Performance & Behavior Notes:
        - **Overhead**: The context manager installs a trace function, which
          introduces performance overhead compared to functional
          usage. Use it for scripts and prototypes rather than hot loops.
        - **Scanning Scope**: The automatic scanning only catches `Err` variants
          that are **assigned to a variable** name in the local scope.
          Unassigned return values will NOT be caught.

    Examples:
        >>> # 1. Functional usage (Low overhead)
        >>> val = assert_ok(Ok(10))
        >>> # assert_ok(Err("fail")) # Raises AssertionError

        >>> # 2. Automatic context manager usage (Higher overhead, Fail-fast)
        >>> with assert_ok("Critical operations"):
        ...     res = Ok(1)  # Fine
        ...     # res2 = Err("boom") # Raises AssertionError immediately

        >>> # 3. Explicit check usage (Lower overhead in context)
        >>> with assert_ok() as ctx:
        ...     val = ctx.check(Ok(42))

    """
    __tracebackhide__ = True
    if isinstance(result_or_message, OkErr):
        match result_or_message:
            case Err(e):  # pyright: ignore[reportUnknownVariableType]
                err_msg = cast("Any", e)
                return _raise_assertion_error(f"assert_ok failed: {err_msg}")
            case Ok(v):  # pyright: ignore[reportUnknownVariableType]
                return cast("Any", v)

    msg = str(result_or_message) if isinstance(result_or_message, str) else "Result was Err"
    return AssertOk(msg)

catch

catch(
    exceptions: type[E],
    func: Callable[P, T],
    *,
    map_to: Any = None,
) -> Callable[P, Result[T, Any]]

catch(
    exceptions: type[E],
    func: Callable[P, Coroutine[Any, Any, T]],
    *,
    map_to: Any = None,
) -> Callable[P, Awaitable[Result[T, Any]]]

catch(
    exceptions: type[E],
    func: None = None,
    *,
    map_to: Any = None,
) -> _CatchDecoratorOrContext[E]

catch(
    exceptions: Mapping[type[Exception], Any],
    func: Callable[P, T],
) -> Callable[P, Result[T, Any]]

catch(
    exceptions: Mapping[type[Exception], Any],
    func: Callable[P, Coroutine[Any, Any, T]],
    *,
    map_to: Any = None,
) -> Callable[P, Awaitable[Result[T, Any]]]

catch(
    exceptions: Mapping[type[Exception], Any],
    func: None = None,
) -> _CatchDecoratorOrContext[Exception]

catch(
    exceptions: tuple[type[E1], type[E2]],
    func: Callable[P, T],
    *,
    map_to: Any = None,
) -> Callable[P, Result[T, Any]]

catch(
    exceptions: tuple[type[E1], type[E2]],
    func: Callable[P, Coroutine[Any, Any, T]],
    *,
    map_to: Any = None,
) -> Callable[P, Awaitable[Result[T, Any]]]

catch(
    exceptions: tuple[type[E1], type[E2]],
    func: None = None,
    *,
    map_to: Any = None,
) -> _CatchDecoratorOrContext[E1 | E2]

catch(
    exceptions: tuple[type[Exception], ...],
    func: Callable[P, T],
    *,
    map_to: Any = None,
) -> Callable[P, Result[T, Any]]

catch(
    exceptions: tuple[type[Exception], ...],
    func: Callable[P, Coroutine[Any, Any, T]],
    *,
    map_to: Any = None,
) -> Callable[P, Awaitable[Result[T, Any]]]

catch(
    exceptions: tuple[type[Exception], ...],
    func: None = None,
    *,
    map_to: Any = None,
) -> _CatchDecoratorOrContext[Exception]

catch(
    exceptions: Any, func: Any = None, *, map_to: Any = None
) -> Any

Execute a function and catch specified exceptions into a Result.

Can be used as a standalone wrapper or as a decorator.

Parameters:

Name	Type	Description	Default
exceptions	Any	An exception type, tuple of types, or mapping of types to values.	required
func	Any	Optional function to wrap and execute immediately.	None
map_to	Any	Optional value to return in Err if an exception matches.	None

Returns:

Type	Description
Any	A Result if func was provided, otherwise a decorator.

Examples:

>>> # 1. Simple catch (returns the caught instance)
>>> @catch(ValueError)
... def parse(s: str) -> int:
...     return int(s)
>>> parse("abc")
Err(ValueError("invalid literal for int()..."))

>>> # 2. Map single error to Enum using map_to
>>> @catch(ValueError, map_to=ErrorCode.INVALID)
... def parse_safe(s: str) -> int:
...     return int(s)
>>> parse_safe("abc")
Err(<ErrorCode.INVALID: 'invalid'>)

>>> # 3. Map multiple errors to Enums using a dictionary
>>> error_map = {ValueError: ErrorCode.INVALID, KeyError: ErrorCode.MISSING}
>>> @catch(error_map)
... def risky_op(x):
...     if x == 0:
...         raise ValueError
...     if x == 1:
...         raise KeyError
...     return "ok"
>>> risky_op(0)
Err(<ErrorCode.INVALID: 'invalid'>)
>>> risky_op(1)
Err(<ErrorCode.MISSING: 'missing'>)

Source code in src/result/result.py

def catch(  # noqa: C901 # pyright: ignore
    exceptions: Any,
    func: Any = None,
    *,
    map_to: Any = None,
) -> Any:
    """Execute a function and catch specified exceptions into a Result.

    Can be used as a standalone wrapper or as a decorator.

    Args:
        exceptions: An exception type, tuple of types, or mapping of types to values.
        func: Optional function to wrap and execute immediately.
        map_to: Optional value to return in Err if an exception matches.

    Returns:
        A Result if `func` was provided, otherwise a decorator.

    Examples:
        >>> # 1. Simple catch (returns the caught instance)
        >>> @catch(ValueError)
        ... def parse(s: str) -> int:
        ...     return int(s)
        >>> parse("abc")
        Err(ValueError("invalid literal for int()..."))

        >>> # 2. Map single error to Enum using map_to
        >>> @catch(ValueError, map_to=ErrorCode.INVALID)
        ... def parse_safe(s: str) -> int:
        ...     return int(s)
        >>> parse_safe("abc")
        Err(<ErrorCode.INVALID: 'invalid'>)

        >>> # 3. Map multiple errors to Enums using a dictionary
        >>> error_map = {ValueError: ErrorCode.INVALID, KeyError: ErrorCode.MISSING}
        >>> @catch(error_map)
        ... def risky_op(x):
        ...     if x == 0:
        ...         raise ValueError
        ...     if x == 1:
        ...         raise KeyError
        ...     return "ok"
        >>> risky_op(0)
        Err(<ErrorCode.INVALID: 'invalid'>)
        >>> risky_op(1)
        Err(<ErrorCode.MISSING: 'missing'>)

    """
    exc_map = _resolve_mapping(exceptions, map_to)
    catch_tuple = tuple(exc_map.keys())
    # Track if we have an actual mapping value (other than the exception class itself)
    has_mapping = map_to is not None or isinstance(exceptions, Mapping)

    def decorator(f: Callable[P, T]) -> Callable[P, Any]:
        if inspect.iscoroutinefunction(f):

            @wraps(f)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[Any, Any]:
                try:
                    return Ok(await f(*args, **kwargs))
                except catch_tuple as e:
                    mapped = exc_map[type(e)] if has_mapping else e
                    return Err(mapped)

            return async_wrapper

        @wraps(f)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> Result[T, Any]:
            __tracebackhide__ = True
            try:
                return Ok(f(*args, **kwargs))
            except catch_tuple as e:
                mapped = exc_map.get(cast("Any", type(e)), e) if has_mapping else e
                return Err(mapped)
            except Exception as e:  # noqa: BLE001
                # Hide the decorator frame from the traceback in modern tools
                # and suppress implementation-detail exception context.
                tb = e.__traceback__
                raise e.with_traceback(tb.tb_next if tb else None) from None

        return wrapper

    if func is not None:
        return decorator(func)

    # If called without a function, check if it's being used as a context manager
    # or a decorator. We return an object that satisfies both.
    class DecoratorOrContext:
        def __init__(self) -> None:
            self._ctx: CatchContext[Any, Any] | None = None

        def __call__(self, f: Callable[P, T]) -> Any:
            return decorator(f)

        def __enter__(self) -> CatchContext[Any, Any]:
            self._ctx = CatchContext(catch_tuple)
            return self._ctx.__enter__()

        def __exit__(
            self,
            exc_type: type[BaseException] | None,
            exc_val: BaseException | None,
            exc_tb: TracebackType | None,
        ) -> bool:
            if self._ctx is None:
                return False
            exit_res = self._ctx.__exit__(exc_type, exc_val, exc_tb)
            if exit_res and self._ctx.result is not None:
                # Apply mapping if context trapped an error
                err_val = self._ctx.result.err()
                if err_val is not None:
                    mapped = exc_map.get(cast("Any", type(err_val)), err_val)
                    self._ctx.result = Err(mapped)
            return exit_res

    return DecoratorOrContext()

catch_call

catch_call(
    exceptions: type[E],
    func: Callable[..., T],
    *args: Any,
    map_to: Any = None,
    **kwargs: Any,
) -> Result[T, Any]

catch_call(
    exceptions: tuple[type[E1], type[E2]],
    func: Callable[..., T],
    *args: Any,
    map_to: Any = None,
    **kwargs: Any,
) -> Result[T, Any]

catch_call(
    exceptions: Mapping[type[Exception], Any],
    func: Callable[..., T],
    *args: Any,
    **kwargs: Any,
) -> Result[T, Any]

catch_call(
    exceptions: tuple[type[Exception], ...],
    func: Callable[..., T],
    *args: Any,
    map_to: Any = None,
    **kwargs: Any,
) -> Result[T, Any]

catch_call(
    exceptions: Any,
    func: Any,
    *args: Any,
    mapping: Mapping[type[Exception], Any] | None = None,
    map_to: Any = None,
    **kwargs: Any,
) -> Any

Execute a function inline and catch specified exceptions into a Result.

This allows for clean, single-line expression evaluation without needing to define a new function or open a context manager block.

Parameters:

Name	Type	Description	Default
exceptions	Any	An exception type, tuple, or mapping to catch.	required
func	Any	The function to execute.	required
*args	Any	Positional arguments for the function.	()
mapping	Mapping[type[Exception], Any] | None	Optional explicit mapping dictionary.	None
map_to	Any	Optional value to use as the error if an exception matches.	None
**kwargs	Any	Keyword arguments for the function.	{}

Returns:

Type	Description
Any	Ok(value) if successful, Err(mapped_error) if a specified exception was raised.

Examples:

>>> import json
>>> # 1. Simple catch
>>> catch_call(json.JSONDecodeError, json.loads, '{"key": "value"}')
Ok({'key': 'value'})

>>> # 2. Catch with mapping
>>> catch_call(json.JSONDecodeError, json.loads, "invalid", map_to="bad_json")
Err('bad_json')

Source code in src/result/result.py

def catch_call(
    exceptions: Any,
    func: Any,
    *args: Any,
    mapping: Mapping[type[Exception], Any] | None = None,
    map_to: Any = None,
    **kwargs: Any,
) -> Any:
    """Execute a function inline and catch specified exceptions into a Result.

    This allows for clean, single-line expression evaluation without needing
    to define a new function or open a context manager block.

    Args:
        exceptions: An exception type, tuple, or mapping to catch.
        func: The function to execute.
        *args: Positional arguments for the function.
        mapping: Optional explicit mapping dictionary.
        map_to: Optional value to use as the error if an exception matches.
        **kwargs: Keyword arguments for the function.

    Returns:
        Ok(value) if successful, Err(mapped_error) if a specified exception was raised.

    Examples:
        >>> import json
        >>> # 1. Simple catch
        >>> catch_call(json.JSONDecodeError, json.loads, '{"key": "value"}')
        Ok({'key': 'value'})

        >>> # 2. Catch with mapping
        >>> catch_call(json.JSONDecodeError, json.loads, "invalid", map_to="bad_json")
        Err('bad_json')

    """
    exc_map = _resolve_mapping(mapping or exceptions, map_to)
    catch_tuple = tuple(exc_map.keys())
    has_mapping = map_to is not None or mapping is not None or isinstance(exceptions, Mapping)

    try:
        return Ok(func(*args, **kwargs))
    except catch_tuple as e:
        mapped = exc_map.get(type(e), e) if has_mapping else e
        return Err(mapped)

combine

combine(
    results: Iterable[Result[T_local, E_local]],
) -> Result[list[T_local], E_local]

Combine an iterable of results into a single result (All-or-Nothing).

If all results are Ok, returns an Ok containing a list of all values. If any result is an Err, returns the first Err encountered.

Parameters:

Name	Type	Description	Default
results	Iterable[Result[T_local, E_local]]	An iterable of Result instances.	required

Returns:

Type	Description
Result[list[T_local], E_local]	A combined Result.

Examples:

>>> combine([Ok(1), Ok(2)])
Ok([1, 2])
>>> combine([Ok(1), Err("fail")])
Err('fail')

Source code in src/result/result.py

def combine[T_local, E_local](results: Iterable[Result[T_local, E_local]]) -> Result[list[T_local], E_local]:
    """Combine an iterable of results into a single result (All-or-Nothing).

    If all results are `Ok`, returns an `Ok` containing a list of all values.
    If any result is an `Err`, returns the first `Err` encountered.

    Args:
        results: An iterable of `Result` instances.

    Returns:
        A combined `Result`.

    Examples:
        >>> combine([Ok(1), Ok(2)])
        Ok([1, 2])
        >>> combine([Ok(1), Err("fail")])
        Err('fail')

    """
    values: list[T_local] = []
    for res in results:
        if isinstance(res, Err):
            return res  # pyright: ignore[reportReturnType]
        values.append(res._value)
    return Ok(values)

do

do(
    gen: Generator[Result[T_co, E_co], Any, T_co],
) -> Result[T_co, E_co]

Helper for inline synchronous do-notation (generator expressions).

If an 'Err' is encountered, it short-circuits and returns that error.

Parameters:

Name	Type	Description	Default
gen	Generator[Result[T_co, E_co], Any, T_co]	A generator expression yielding Results.	required

Returns:

Type	Description
Result[T_co, E_co]	The final Result of the chain.

Examples:

>>> do(Ok(x + y) for x in Ok(1) for y in Ok(2))
Ok(3)

Source code in src/result/result.py

def do[T_co, E_co](gen: Generator[Result[T_co, E_co], Any, T_co]) -> Result[T_co, E_co]:
    """Helper for inline synchronous do-notation (generator expressions).

    If an 'Err' is encountered, it short-circuits and returns that error.

    Args:
        gen: A generator expression yielding Results.

    Returns:
        The final Result of the chain.

    Examples:
        >>> do(Ok(x + y) for x in Ok(1) for y in Ok(2))
        Ok(3)

    """
    try:
        return next(gen)
    except _DoError as e:
        return e.err
    except StopIteration as e:
        # If the generator finishes normally, wrap its return value in Ok
        return Ok(e.value)

do_async async

do_async(
    gen: AsyncGenerator[Result[T_co, E_co], Any],
) -> Result[T_co, E_co]

Helper for inline asynchronous do-notation (generator expressions).

If an 'Err' is encountered, it short-circuits and returns that error.

Parameters:

Name	Type	Description	Default
gen	AsyncGenerator[Result[T_co, E_co], Any]	An async generator expression yielding Results.	required

Returns:

Type	Description
Result[T_co, E_co]	The final Result of the chain.

Examples:

>>> async def async_gen():
...     yield Ok(1)
>>> await do_async(Ok(x + 2) async for x in async_gen())
Ok(3)

Source code in src/result/result.py

async def do_async[T_co, E_co](gen: AsyncGenerator[Result[T_co, E_co], Any]) -> Result[T_co, E_co]:
    """Helper for inline asynchronous do-notation (generator expressions).

    If an 'Err' is encountered, it short-circuits and returns that error.

    Args:
        gen: An async generator expression yielding Results.

    Returns:
        The final Result of the chain.

    Examples:
        >>> async def async_gen():
        ...     yield Ok(1)
        >>> await do_async(Ok(x + 2) async for x in async_gen())
        Ok(3)

    """
    try:
        return await anext(gen)
    except _DoError as e:
        return e.err
    except StopAsyncIteration:
        msg = "Async do-notation generator ended without yielding a Result"
        raise UnwrapError(Ok(None), msg) from None

do_notation

do_notation(
    arg: Callable[P, Do[T_local, E_local]],
    *,
    catch: None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Callable[P, Result[T_local, E_local]]

do_notation(
    arg: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    *,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Callable[
    [Callable[P, Do[T_local, E_local]]],
    Callable[P, Result[T_local, E_local | Exception]],
]

do_notation(
    arg: Callable[P, Do[T_local, E_local]]
    | type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    *,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Any

Enable imperative-style 'do-notation' for synchronous Result blocks.

This decorator allows writing code that looks procedural by using yield to unwrap Result values. If any yield receives an Err, the function short-circuits immediately and returns that Err.

Notes

The final 'return' value is automatically wrapped in an Ok variant.

Parameters:

Name	Type	Description	Default
arg	Callable[P, Do[T_local, E_local]] | type[Exception] | tuple[type[Exception], ...] | None	The function to decorate, or an exception type to catch.	None
catch	type[Exception] | tuple[type[Exception], ...] | None	Optional keyword-only exception type to catch and lift into Err.	None
remap	dict[type[Any], type[Any]] | None	Optional mapping of internal error types to high-level domain errors.	None

Returns:

Type	Description
Any	The decorated function or a decorator factory.

Examples:

>>> @do_notation
... def process(x):
...     a = yield Ok(x + 1)
...     b = yield Ok(a * 2)
...     return b
>>> process(5)
Ok(12)

>>> @do_notation(catch=ValueError)
... def risky(s):
...     val = yield Ok(int(s))
...     return val
>>> risky("not a number")
Err(ValueError(...))

Source code in src/result/result.py

def do_notation[T_local, E_local, **P](
    arg: Callable[P, Do[T_local, E_local]] | type[Exception] | tuple[type[Exception], ...] | None = None,
    *,
    catch: type[Exception] | tuple[type[Exception], ...] | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Any:
    """Enable imperative-style 'do-notation' for synchronous Result blocks.

    This decorator allows writing code that looks procedural by using `yield`
    to unwrap `Result` values. If any `yield` receives an `Err`, the function
    short-circuits immediately and returns that Err.

    Notes:
        The final 'return' value is automatically wrapped in an Ok variant.

    Args:
        arg: The function to decorate, or an exception type to catch.
        catch: Optional keyword-only exception type to catch and lift into Err.
        remap: Optional mapping of internal error types to high-level domain errors.

    Returns:
        The decorated function or a decorator factory.

    Examples:
        >>> @do_notation
        ... def process(x):
        ...     a = yield Ok(x + 1)
        ...     b = yield Ok(a * 2)
        ...     return b
        >>> process(5)
        Ok(12)

        >>> @do_notation(catch=ValueError)
        ... def risky(s):
        ...     val = yield Ok(int(s))
        ...     return val
        >>> risky("not a number")
        Err(ValueError(...))

    """
    if callable(arg) and not isinstance(arg, type | tuple):
        return _make_do_wrapper(arg, None, remap)
    catch_final = catch or (arg if isinstance(arg, type | tuple) else None)  # pyright: ignore[reportUnknownVariableType]

    def decorator(func: Callable[P, Do[T_local, E_local]]) -> Callable[P, Result[T_local, E_local | Exception]]:
        return _make_do_wrapper(func, cast("Any", catch_final), remap)

    return decorator

do_notation_async

do_notation_async(
    arg: Callable[P, DoAsync[T_local, E_local]],
    *,
    catch: None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Callable[
    P, Coroutine[Any, Any, Result[T_local, E_local]]
]

do_notation_async(
    arg: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    *,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Callable[
    [Callable[P, DoAsync[T_local, E_local]]],
    Callable[
        P,
        Coroutine[
            Any, Any, Result[T_local, E_local | Exception]
        ],
    ],
]

do_notation_async(
    arg: Callable[P, DoAsync[T_local, E_local]]
    | type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    *,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Any

Enable imperative-style 'do-notation' for asynchronous Result blocks.

This is the async version of @do_notation. It supports yield await for asynchronous Result-returning functions.

Notes

Unlike the sync version, the final result MUST be yielded as an Ok variant because async generators do not support native return values in this context.

Parameters:

Name	Type	Description	Default
arg	Callable[P, DoAsync[T_local, E_local]] | type[Exception] | tuple[type[Exception], ...] | None	The async function to decorate, or an exception type to catch.	None
catch	type[Exception] | tuple[type[Exception], ...] | None	Optional keyword-only exception type to catch and lift into Err.	None
remap	dict[type[Any], type[Any]] | None	Optional mapping of internal error types to high-level domain errors.	None

Returns:

Type	Description
Any	The decorated async function or a decorator factory.

Examples:

>>> @do_notation_async
... async def get_data(user_id):
...     user = yield await fetch_user(user_id)  # fetch_user returns Result
...     data = yield await fetch_posts(user["id"])
...     yield Ok(data)
>>> await get_data(1)
Ok([...])

Source code in src/result/result.py

def do_notation_async[T_local, E_local, **P](
    arg: Callable[P, DoAsync[T_local, E_local]] | type[Exception] | tuple[type[Exception], ...] | None = None,
    *,
    catch: type[Exception] | tuple[type[Exception], ...] | None = None,
    remap: dict[type[Any], type[Any]] | None = None,
) -> Any:
    """Enable imperative-style 'do-notation' for asynchronous Result blocks.

    This is the async version of `@do_notation`. It supports `yield await` for
    asynchronous Result-returning functions.

    Notes:
        Unlike the sync version, the final result MUST be yielded as an Ok
        variant because async generators do not support native return values
        in this context.

    Args:
        arg: The async function to decorate, or an exception type to catch.
        catch: Optional keyword-only exception type to catch and lift into Err.
        remap: Optional mapping of internal error types to high-level domain errors.

    Returns:
        The decorated async function or a decorator factory.

    Examples:
        >>> @do_notation_async
        ... async def get_data(user_id):
        ...     user = yield await fetch_user(user_id)  # fetch_user returns Result
        ...     data = yield await fetch_posts(user["id"])
        ...     yield Ok(data)
        >>> await get_data(1)
        Ok([...])

    """
    if callable(arg) and not isinstance(arg, type | tuple):
        return _make_async_wrapper(arg, None, remap)
    catch_final = catch or (arg if isinstance(arg, type | tuple) else None)  # pyright: ignore[reportUnknownVariableType]

    def decorator(func: Callable[P, DoAsync[T_local, E_local]]) -> Callable[P, Coroutine[Any, Any, Result[Any, Any]]]:
        return _make_async_wrapper(func, cast("Any", catch_final), remap)

    return decorator

from_optional

from_optional(
    value: T_co | None, error: E_co
) -> Result[T_co, E_co]

Create a Result from an optional value.

Parameters:

Name	Type	Description	Default
value	T_co | None	The optional value to wrap.	required
error	E_co	The error to return if the value is None.	required

Returns:

Type	Description
Result[T_co, E_co]	Ok(value) if value is not None, otherwise Err(error).

Examples:

>>> from_optional(42, "fail")
Ok(42)
>>> from_optional(None, "fail")
Err('fail')

Source code in src/result/result.py

def from_optional[T_co, E_co](value: T_co | None, error: E_co) -> Result[T_co, E_co]:
    """Create a Result from an optional value.

    Args:
        value: The optional value to wrap.
        error: The error to return if the value is None.

    Returns:
        Ok(value) if value is not None, otherwise Err(error).

    Examples:
        >>> from_optional(42, "fail")
        Ok(42)
        >>> from_optional(None, "fail")
        Err('fail')

    """
    if value is None:
        return Err(error)
    return Ok(value)

is_err

is_err(
    result: Result[T_local, E_local],
) -> TypeIs[Err[E_local]]

Check if a result is an Err variant and narrow its type.

Examples:

>>> res: Result[int, str] = Err("fail")
>>> if is_err(res):
...     # res is now typed as Err[str]
...     print(res._error)

Source code in src/result/result.py

def is_err[T_local, E_local](result: Result[T_local, E_local]) -> TypeIs[Err[E_local]]:
    """Check if a result is an `Err` variant and narrow its type.

    Examples:
        >>> res: Result[int, str] = Err("fail")
        >>> if is_err(res):
        ...     # res is now typed as Err[str]
        ...     print(res._error)

    """
    return isinstance(result, Err)

is_ok

is_ok(
    result: Result[T_local, E_local],
) -> TypeIs[Ok[T_local]]

Check if a result is an Ok variant and narrow its type.

Examples:

>>> res: Result[int, str] = Ok(10)
>>> if is_ok(res):
...     # res is now typed as Ok[int]
...     print(res._value)

Source code in src/result/result.py

def is_ok[T_local, E_local](result: Result[T_local, E_local]) -> TypeIs[Ok[T_local]]:
    """Check if a result is an `Ok` variant and narrow its type.

    Examples:
        >>> res: Result[int, str] = Ok(10)
        >>> if is_ok(res):
        ...     # res is now typed as Ok[int]
        ...     print(res._value)

    """
    return isinstance(result, Ok)

map2

map2(
    res1: Result[T1, E1],
    res2: Result[T2, E2],
    func: Callable[[T1, T2], U],
) -> Result[U, E1 | E2]

Apply a binary function to the values of two results (Zipping Map).

If both are Ok, returns Ok(func(val1, val2)). If either is Err, returns the first Err encountered.

Parameters:

Name	Type	Description	Default
res1	Result[T1, E1]	First Result.	required
res2	Result[T2, E2]	Second Result.	required
func	Callable[[T1, T2], U]	A function taking two success values.	required

Returns:

Type	Description
Result[U, E1 | E2]	A new Result.

Examples:

>>> map2(Ok(1), Ok(2), lambda x, y: x + y)
Ok(3)

Source code in src/result/result.py

def map2[T1, T2, U, E1, E2](
    res1: Result[T1, E1],
    res2: Result[T2, E2],
    func: Callable[[T1, T2], U],
) -> Result[U, E1 | E2]:
    """Apply a binary function to the values of two results (Zipping Map).

    If both are Ok, returns Ok(func(val1, val2)).
    If either is Err, returns the first Err encountered.

    Args:
        res1: First Result.
        res2: Second Result.
        func: A function taking two success values.

    Returns:
        A new Result.

    Examples:
        >>> map2(Ok(1), Ok(2), lambda x, y: x + y)
        Ok(3)

    """
    if isinstance(res1, Err):
        return cast("Result[U, E1 | E2]", res1)
    if isinstance(res2, Err):
        return cast("Result[U, E1 | E2]", res2)
    return Ok(func(res1._value, res2._value))

partition

partition(
    results: Iterable[Result[T_local, E_local]],
) -> tuple[list[T_local], list[E_local]]

Partition an iterable of results into two lists.

Unlike combine, this does not short-circuit. It collects all values into two separate lists.

Parameters:

Name	Type	Description	Default
results	Iterable[Result[T_local, E_local]]	An iterable of Result instances.	required

Returns:

Type	Description
tuple[list[T_local], list[E_local]]	A tuple of (ok_values, err_states).

Examples:

>>> partition([Ok(1), Err("a"), Ok(2)])
([1, 2], ['a'])

Source code in src/result/result.py

def partition[T_local, E_local](results: Iterable[Result[T_local, E_local]]) -> tuple[list[T_local], list[E_local]]:
    """Partition an iterable of results into two lists.

    Unlike `combine`, this does not short-circuit. It collects all values
    into two separate lists.

    Args:
        results: An iterable of `Result` instances.

    Returns:
        A tuple of (ok_values, err_states).

    Examples:
        >>> partition([Ok(1), Err("a"), Ok(2)])
        ([1, 2], ['a'])

    """
    oks: list[T_local] = []
    errs: list[E_local] = []
    for res in results:
        if isinstance(res, Ok):
            oks.append(res._value)
        else:
            errs.append(res._error)
    return oks, errs

retry_result

retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[E], bool],
    catch: None = None,
) -> Callable[
    [Callable[P, Result[T, E]]],
    Callable[P, Result[T, E]],
]

retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: None = None,
    catch: None = None,
) -> Callable[
    [Callable[P, Result[T, E]]],
    Callable[P, Result[T, E]],
]

retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[E_exc], bool] | None = None,
    catch: type[E_exc],
) -> Callable[
    [Callable[P, T]], Callable[P, Result[T, E_exc]]
]

retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    catch: tuple[type[Exception], ...]
    | Mapping[type[Exception], Any],
) -> Callable[
    [Callable[P, T | Result[T, Any]]],
    Callable[P, Result[T, Any]],
]

retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[Any], bool] | None = None,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | Mapping[type[Exception], Any]
    | None = None,
) -> Any

A resilience decorator for synchronous functions that return a Result.

It will automatically re-execute the function if it returns an Err variant, up to the specified number of attempts.

If catch is provided, it will also catch specified exceptions and turn them into Err variants before deciding whether to retry.

Parameters:

Name	Type	Description	Default
attempts	int	Total number of attempts to try (default 3).	3
delay	float	Initial delay in seconds between retries (default 0).	0
backoff	float	Multiplier for the delay after each attempt (default 1.0).	1.0
jitter	bool | float	If True (or a float), add randomness to the delay.	False
retry_if	Callable[[Any], bool] | None	Optional predicate to decide if an error should be retried.	None
catch	type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any] | None	Optional exception types to catch and lift into Results.	None

Returns:

Type	Description
Any	A decorator that adds retry logic to the function.

Notes & Footguns: - Idempotency: Retrying functions with side effects (e.g., DB writes) can be dangerous if the operation is not idempotent. - Exception Scoping: If catch is NOT used, and the function raises an exception, the retry logic will NOT trigger (the exception will bubble up). The retry logic only reacts to Err return values. - Wait Times: High attempts and backoff values can lead to extremely long execution times.

Examples:

>>> # 1. Basic Retry (Reacts to Err return)
>>> @retry_result(attempts=3)
... def unstable():
...     return Err("fail")
>>> unstable()
Err('fail')  # Tried 3 times

>>> # 2. Exponential Backoff with Jitter
>>> @retry_result(attempts=5, delay=0.1, backoff=2.0, jitter=True)
... def network_call():
...     return Err("timeout")

>>> # 3. Conditional Retry (only retry on transient errors)
>>> @retry_result(attempts=3, retry_if=lambda e: e == "temporary")
... def db_op():
...     return Err("permanent")
>>> db_op()
Err('permanent')  # Fails fast, only tried once

>>> # 4. Internal Exception Catching (Lifting)
>>> @retry_result(attempts=3, catch=ValueError)
... def parse_stuff(s):
...     return int(s)  # Raises ValueError -> Err -> Retry
>>> parse_stuff("abc")
Err(ValueError("invalid literal for int()..."))

>>> # 5. Stacking with @catch
>>> @retry_result(attempts=2)
... @catch(KeyError)
... def get_config():
...     raise KeyError("missing")
>>> get_config()
Err(KeyError('missing'))  # @catch fires first, then @retry sees Err

Source code in src/result/result.py

def retry_result(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[Any], bool] | None = None,
    catch: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any] | None = None,
) -> Any:
    """A resilience decorator for synchronous functions that return a `Result`.

    It will automatically re-execute the function if it returns an `Err` variant,
    up to the specified number of `attempts`.

    If `catch` is provided, it will also catch specified exceptions and turn
    them into `Err` variants before deciding whether to retry.

    Args:
        attempts: Total number of attempts to try (default 3).
        delay: Initial delay in seconds between retries (default 0).
        backoff: Multiplier for the delay after each attempt (default 1.0).
        jitter: If True (or a float), add randomness to the delay.
        retry_if: Optional predicate to decide if an error should be retried.
        catch: Optional exception types to catch and lift into Results.

    Returns:
        A decorator that adds retry logic to the function.

    Notes & Footguns:
        - **Idempotency**: Retrying functions with side effects (e.g., DB writes)
          can be dangerous if the operation is not idempotent.
        - **Exception Scoping**: If `catch` is NOT used, and the function raises
          an exception, the retry logic will NOT trigger (the exception will
          bubble up). The retry logic only reacts to `Err` return values.
        - **Wait Times**: High `attempts` and `backoff` values can lead to
          extremely long execution times.

    Examples:
        >>> # 1. Basic Retry (Reacts to Err return)
        >>> @retry_result(attempts=3)
        ... def unstable():
        ...     return Err("fail")
        >>> unstable()
        Err('fail')  # Tried 3 times

        >>> # 2. Exponential Backoff with Jitter
        >>> @retry_result(attempts=5, delay=0.1, backoff=2.0, jitter=True)
        ... def network_call():
        ...     return Err("timeout")

        >>> # 3. Conditional Retry (only retry on transient errors)
        >>> @retry_result(attempts=3, retry_if=lambda e: e == "temporary")
        ... def db_op():
        ...     return Err("permanent")
        >>> db_op()
        Err('permanent')  # Fails fast, only tried once

        >>> # 4. Internal Exception Catching (Lifting)
        >>> @retry_result(attempts=3, catch=ValueError)
        ... def parse_stuff(s):
        ...     return int(s)  # Raises ValueError -> Err -> Retry
        >>> parse_stuff("abc")
        Err(ValueError("invalid literal for int()..."))

        >>> # 5. Stacking with @catch
        >>> @retry_result(attempts=2)
        ... @catch(KeyError)
        ... def get_config():
        ...     raise KeyError("missing")
        >>> get_config()
        Err(KeyError('missing'))  # @catch fires first, then @retry sees Err

    """

    def decorator[T, E, **P](f: Callable[P, Result[T, E] | T]) -> Callable[P, Result[T, E]]:
        actual_catch = globals()["catch"]
        wrapped_f = actual_catch(cast("Any", catch_val))(f) if (catch_val := catch) else f

        @wraps(f)
        def wrapper(*args: Any, **kwargs: Any) -> Result[T, E]:
            current_delay = delay
            # Initialize with dummy error, will be overwritten in loop
            res: Any = Err(cast("E", "retry loop didn't run"))  # pyright: ignore[reportUnknownVariableType]
            for i in range(attempts):
                # Call the (potentially catch-wrapped) function
                res = wrapped_f(*args, **kwargs)

                # If it's not a Result (e.g. catch wasn't used and func returns T),
                # we wrap it in Ok automatically.
                if not isinstance(res, OkErr):
                    res = Ok(res)

                if isinstance(res, Ok):
                    return res  # pyright: ignore[reportUnknownVariableType]

                # It's an Err, check if we should retry
                err_val = res.err()  # pyright: ignore[reportUnknownVariableType]
                if retry_if and not retry_if(err_val):
                    return res  # pyright: ignore[reportUnknownVariableType]

                # Final attempt failed
                if i == attempts - 1:
                    return res  # pyright: ignore[reportUnknownVariableType]

                # Sleep and backoff
                sleep_time = _get_retry_delay(current_delay, jitter=jitter)
                if sleep_time > 0:
                    time.sleep(sleep_time)

                current_delay *= backoff
            return res  # pyright: ignore[reportUnknownVariableType]

        return cast("Any", wrapper)

    return cast("Any", decorator)

retry_result_async

retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[E], bool],
    catch: None = None,
) -> Callable[
    [Callable[P, Awaitable[Result[T, E]]]],
    Callable[P, Awaitable[Result[T, E]]],
]

retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: None = None,
    catch: None = None,
) -> Callable[
    [Callable[P, Awaitable[Result[T, E]]]],
    Callable[P, Awaitable[Result[T, E]]],
]

retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[E_exc], bool] | None = None,
    catch: type[E_exc],
) -> Callable[
    [Callable[P, Awaitable[T]]],
    Callable[P, Awaitable[Result[T, E_exc]]],
]

retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    catch: tuple[type[Exception], ...]
    | Mapping[type[Exception], Any],
) -> Callable[
    [
        Callable[
            P, Awaitable[T] | Awaitable[Result[T, Any]]
        ]
    ],
    Callable[P, Awaitable[Result[T, Any]]],
]

retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[Any], bool] | None = None,
    catch: type[Exception]
    | tuple[type[Exception], ...]
    | Mapping[type[Exception], Any]
    | None = None,
) -> Any

A resilience decorator for asynchronous functions that return a Result.

This is the asynchronous version of retry_result.

It will automatically re-execute the function if it returns an Err variant, up to the specified number of attempts.

If catch is provided, it will also catch specified exceptions and turn them into Err variants before deciding whether to retry.

Parameters:

Name	Type	Description	Default
attempts	int	Total number of attempts to try (default 3).	3
delay	float	Initial delay in seconds between retries (default 0).	0
backoff	float	Multiplier for the delay after each attempt (default 1.0).	1.0
jitter	bool | float	If True (or a float), add randomness to the delay.	False
retry_if	Callable[[Any], bool] | None	Optional predicate to decide if an error should be retried.	None
catch	type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any] | None	Optional exception types to catch and lift into Results.	None

Returns:

Type	Description
Any	A decorator that adds retry logic to the async function.

Notes & Footguns: - Idempotency: Retrying functions with side effects (e.g., POST calls) can be dangerous if the operation is not idempotent. - Exception Scoping: If catch is NOT used, and the function raises an exception, the retry logic will NOT trigger (the exception will bubble up). - Concurrency: This decorator retries in series. For parallel retries, consider other orchestration patterns.

Examples:

>>> # 1. Async Basic Retry
>>> @retry_result_async(attempts=3)
... async def unstable():
...     return Err("fail")
>>> await unstable()
Err('fail')

>>> # 2. Async Backoff with Jitter
>>> @retry_result_async(attempts=5, delay=0.1, backoff=2.0, jitter=0.5)
... async def fetch_data():
...     return Err("timeout")

>>> # 3. Async Conditional Retry
>>> @retry_result_async(attempts=3, retry_if=lambda e: e.status == 429)
... async def api_call():
...     return Err(Response(status=500))
>>> await api_call()
Err(Response(status=500))  # Fails fast, not a 429

>>> # 4. Async Internal Exception Catching
>>> @retry_result_async(attempts=3, catch=asyncio.TimeoutError)
... async def timed_op():
...     raise asyncio.TimeoutError
>>> await timed_op()
Err(asyncio.TimeoutError())

Source code in src/result/result.py

def retry_result_async(
    attempts: int = 3,
    delay: float = 0,
    backoff: float = 1.0,
    *,
    jitter: bool | float = False,
    retry_if: Callable[[Any], bool] | None = None,
    catch: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any] | None = None,
) -> Any:
    """A resilience decorator for asynchronous functions that return a `Result`.

    This is the asynchronous version of `retry_result`.

    It will automatically re-execute the function if it returns an `Err` variant,
    up to the specified number of `attempts`.

    If `catch` is provided, it will also catch specified exceptions and turn
    them into `Err` variants before deciding whether to retry.

    Args:
        attempts: Total number of attempts to try (default 3).
        delay: Initial delay in seconds between retries (default 0).
        backoff: Multiplier for the delay after each attempt (default 1.0).
        jitter: If True (or a float), add randomness to the delay.
        retry_if: Optional predicate to decide if an error should be retried.
        catch: Optional exception types to catch and lift into Results.

    Returns:
        A decorator that adds retry logic to the async function.

    Notes & Footguns:
        - **Idempotency**: Retrying functions with side effects (e.g., POST calls)
          can be dangerous if the operation is not idempotent.
        - **Exception Scoping**: If `catch` is NOT used, and the function raises
          an exception, the retry logic will NOT trigger (the exception will
          bubble up).
        - **Concurrency**: This decorator retries in series. For parallel
          retries, consider other orchestration patterns.

    Examples:
        >>> # 1. Async Basic Retry
        >>> @retry_result_async(attempts=3)
        ... async def unstable():
        ...     return Err("fail")
        >>> await unstable()
        Err('fail')

        >>> # 2. Async Backoff with Jitter
        >>> @retry_result_async(attempts=5, delay=0.1, backoff=2.0, jitter=0.5)
        ... async def fetch_data():
        ...     return Err("timeout")

        >>> # 3. Async Conditional Retry
        >>> @retry_result_async(attempts=3, retry_if=lambda e: e.status == 429)
        ... async def api_call():
        ...     return Err(Response(status=500))
        >>> await api_call()
        Err(Response(status=500))  # Fails fast, not a 429

        >>> # 4. Async Internal Exception Catching
        >>> @retry_result_async(attempts=3, catch=asyncio.TimeoutError)
        ... async def timed_op():
        ...     raise asyncio.TimeoutError
        >>> await timed_op()
        Err(asyncio.TimeoutError())

    """

    def decorator[T, E, **P](
        f: Callable[P, Awaitable[Result[T, E]] | Awaitable[T]],
    ) -> Callable[P, Awaitable[Result[T, E]]]:
        # If catch is provided, we wrap the function in @catch first
        # We must ensure the resulting wrapper is awaited
        actual_catch = globals()["catch"]
        wrapped_f = actual_catch(cast("Any", catch_val))(f) if (catch_val := catch) else f

        @wraps(f)
        async def wrapper(*args: Any, **kwargs: Any) -> Result[T, E]:
            current_delay = delay
            res: Any = Err(cast("E", "retry loop didn't run"))  # pyright: ignore[reportUnknownVariableType]
            for i in range(attempts):
                res = await wrapped_f(*args, **kwargs)  # pyright: ignore[reportUnknownVariableType, reportGeneralTypeIssues]

                if not isinstance(res, OkErr):
                    res = Ok(res)  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]

                if isinstance(res, Ok):
                    return res  # pyright: ignore[reportUnknownVariableType, reportUnknownVariableType]

                err_val = res.err()  # pyright: ignore[reportUnknownVariableType]
                if retry_if and not retry_if(err_val):
                    return res  # pyright: ignore[reportUnknownVariableType]

                if i == attempts - 1:
                    return res  # pyright: ignore[reportUnknownVariableType]

                sleep_time = _get_retry_delay(current_delay, jitter=jitter)
                if sleep_time > 0:
                    await asyncio.sleep(sleep_time)

                current_delay *= backoff
            return res  # pyright: ignore[reportUnknownVariableType]

        return cast("Any", wrapper)

    return cast("Any", decorator)