Adapters

Bulk lifting and integration utilities for the Result Pattern.

result.adapters

Adapters: Bulk lifting and integration utilities for the Result Pattern.

This module provides tools to bridge the gap between third-party imperative APIs and the functional world of Results. This includes class-level decorators, instance proxies, and fault-tolerant iteration.

Note

Most utilities in this module use dynamic proxying or bulk decoration, which can lead to Type Erasure in some static analysis tools.

collecting module-attribute

collecting = _CollectingFactory()

Utility for collecting errors from multiple exception handlers.

Examples:

>>> with collecting() as col:
...     name = col.add(parse_name(raw_name))  # Returns value if Ok, else None
...     age = col.add(parse_age(raw_age))
...     email = col.add(parse_email(raw_email))
...
... if col.ok:
...     return Ok(User(name, age, email))
... return Err(col.errors)  # List of all failures

Collector

Collector()

Imperative error accumulator for non-short-circuiting logic.

Source code in src/result/adapters.py

def __init__(self) -> None:
    self._errors: list[E_acc] = []

errors property

errors: list[E_acc]

Return the list of all collected errors.

ok property

ok: bool

Check if any errors were collected.

add

add(result: Result[T, E_acc]) -> T | None

Add a result to the collector.

Parameters:

Name	Type	Description	Default
result	Result[T, E_acc]	The Result to add.	required

Returns:

Type	Description
T | None	The success value if Ok, otherwise None.

Source code in src/result/adapters.py

def add[T](self, result: Result[T, E_acc]) -> T | None:
    """Add a result to the collector.

    Args:
        result: The Result to add.

    Returns:
        The success value if Ok, otherwise None.

    """
    match result:
        case Ok(v):
            return v
        case Err(e):
            self._errors.append(e)
            return None

SafeStream

SafeStream(gen: Iterator[Result[T, E]])

Bases: Iterable['Result[T, E]']

A wrapper around a fallible generator that provides functional transposition.

SafeStream captures exceptions during iteration and converts them into Err variants. It provides methods to transpose the entire stream into a single Result or Outcome.

Note

Like generators, a SafeStream can only be iterated once.

Initialize a SafeStream with a Result-yielding iterator.

Source code in src/result/adapters.py

def __init__(self, gen: Iterator[Result[T, E]]) -> None:
    """Initialize a SafeStream with a Result-yielding iterator."""
    self._gen = gen
    self._consumed = False

__iter__

__iter__() -> Iterator[Result[T, E]]

Iterate over the stream.

Raises:

Type	Description
RuntimeError	If the stream is iterated more than once.

Source code in src/result/adapters.py

def __iter__(self) -> Iterator[Result[T, E]]:
    """Iterate over the stream.

    Raises:
        RuntimeError: If the stream is iterated more than once.

    """
    if self._consumed:
        msg = "SafeStream can only be iterated once"
        raise RuntimeError(msg)
    self._consumed = True
    yield from self._gen

to_outcome

to_outcome() -> Outcome[list[T], list[E]]

Transpose the stream into a fault-tolerant Outcome (Partial Success).

Collects all success values and all error values into a single Outcome.

Source code in src/result/adapters.py

def to_outcome(self) -> Outcome[list[T], list[E]]:
    """Transpose the stream into a fault-tolerant Outcome (Partial Success).

    Collects all success values and all error values into a single Outcome.

    """
    oks, errs = partition(list(self))
    try:
        from .outcome import Outcome  # noqa: PLC0415
    except ImportError:
        raise ImportError(
            r"Outcome is not available. use the \`result_pattern\` pip package and not just the result.py file"
        ) from None

    return Outcome(oks, errs or None)

to_result

to_result() -> Result[list[T], E]

Transpose the stream into a single Result (All-or-Nothing).

If any item in the stream is an Err, the first Err encountered is returned. Otherwise, returns Ok(list) containing all success values.

Source code in src/result/adapters.py

def to_result(self) -> Result[list[T], E]:
    """Transpose the stream into a single Result (All-or-Nothing).

    If any item in the stream is an Err, the first Err encountered is returned.
    Otherwise, returns Ok(list) containing all success values.

    """
    return combine(list(self))

SafeStreamAsync

SafeStreamAsync(gen: AsyncIterator[Result[T, E]])

Bases: AsyncIterable['Result[T, E]']

Async version of SafeStream for fallible asynchronous generators.

Initialize a SafeStreamAsync with a Result-yielding async iterator.

Source code in src/result/adapters.py

def __init__(self, gen: AsyncIterator[Result[T, E]]) -> None:
    """Initialize a SafeStreamAsync with a Result-yielding async iterator."""
    self._gen = gen
    self._consumed = False

__aiter__ async

__aiter__() -> AsyncIterator[Result[T, E]]

Iterate over the async stream.

Raises:

Type	Description
RuntimeError	If the stream is iterated more than once.

Source code in src/result/adapters.py

async def __aiter__(self) -> AsyncIterator[Result[T, E]]:
    """Iterate over the async stream.

    Raises:
        RuntimeError: If the stream is iterated more than once.

    """
    if self._consumed:
        msg = "SafeStreamAsync can only be iterated once"
        raise RuntimeError(msg)
    self._consumed = True
    async for item in self._gen:
        yield item

to_outcome async

to_outcome() -> Outcome[list[T], list[E]]

Transpose the async stream into a fault-tolerant Outcome (Partial Success).

Source code in src/result/adapters.py

async def to_outcome(self) -> Outcome[list[T], list[E]]:
    """Transpose the async stream into a fault-tolerant Outcome (Partial Success)."""
    items = [res async for res in self]
    oks, errs = partition(items)
    try:
        from .outcome import Outcome  # noqa: PLC0415
    except ImportError:
        raise ImportError(
            r"Outcome is not available. use the \`result_pattern\` pip package and not just the result.py file"
        ) from None

    return Outcome(oks, errs or None)

to_result async

to_result() -> Result[list[T], E]

Transpose the async stream into a single Result (All-or-Nothing).

Source code in src/result/adapters.py

async def to_result(self) -> Result[list[T], E]:
    """Transpose the async stream into a single Result (All-or-Nothing)."""
    items = [res async for res in self]
    return combine(items)

ScopedCatch

ScopedCatch()

Multi-exception type routing context manager.

Source code in src/result/adapters.py

def __init__(self) -> None:
    self._handlers: dict[type[Exception], Any] = {}
    self.result: Result[T, E] | None = None

on

on(exc_type: type[Exception], map_to: E_new) -> None

Register a mapping for a specific exception type.

Source code in src/result/adapters.py

def on[E_new](self, exc_type: type[Exception], map_to: E_new) -> None:  # pyright: ignore[reportInvalidTypeVarUse]
    """Register a mapping for a specific exception type."""
    self._handlers[exc_type] = map_to

set

set(value: T) -> None

Set the success result for the scope.

Source code in src/result/adapters.py

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

catch_boundary

catch_boundary(
    exceptions: type[Exception]
    | tuple[type[Exception], ...]
    | Mapping[type[Exception], Any],
    *,
    map_to: Any = None,
) -> Callable[[T_cls], T_cls]

Wrap all public methods of a class with the @catch decorator.

This is an 'Entry Adapter' that allows lifting an entire external SDK or client class into the Result world in a single declaration.

Parameters:

Name	Type	Description	Default
exceptions	type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any]	The exceptions to catch on all methods.	required
map_to	Any	Optional constant error value.	None

Returns:

Type	Description
Callable[[T_cls], T_cls]	A class decorator.

Static Analysis Note (Type Erasure): Using this decorator causes Type Erasure. Most Python type checkers (Mypy, Pyright) cannot currently track that the return types of all methods have been transformed from T to Result[T, E].

Source code in src/result/adapters.py

def catch_boundary(
    exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any],
    *,
    map_to: Any = None,
) -> Callable[[T_cls], T_cls]:
    """Wrap all public methods of a class with the @catch decorator.

    This is an 'Entry Adapter' that allows lifting an entire external SDK or
    client class into the Result world in a single declaration.

    Args:
        exceptions: The exceptions to catch on all methods.
        map_to: Optional constant error value.

    Returns:
        A class decorator.

    Static Analysis Note (Type Erasure):
        Using this decorator causes **Type Erasure**. Most Python type checkers
        (Mypy, Pyright) cannot currently track that the return types of all
        methods have been transformed from `T` to `Result[T, E]`.

    """

    def decorator(cls: T_cls) -> T_cls:
        for name, method in inspect.getmembers(cls, predicate=inspect.isroutine):
            if name.startswith("_"):
                continue
            # Wrap the method with @catch using original parameters
            # Use Any to satisfy ty's overload resolution
            setattr(cls, name, catch(cast("Any", exceptions), map_to=map_to)(method))
        return cls

    return decorator

catch_each_iter

catch_each_iter(
    exceptions: type[E_local]
    | tuple[type[E_local], ...]
    | Mapping[type[E_local], Any],
    *,
    map_to: Any = None,
) -> Callable[
    [Callable[P_local, Iterator[T_local]]],
    Callable[P_local, SafeStream[T_local, Any]],
]

Wrap a generator function to capture iteration-level exceptions into a SafeStream.

This decorator ensures that if an exception is raised during the iteration of the generator, it is caught and yielded as an Err variant.

Parameters:

Name	Type	Description	Default
exceptions	type[E_local] | tuple[type[E_local], ...] | Mapping[type[E_local], Any]	One or more exception types to catch, or a mapping of exception types to error values.	required
map_to	Any	Optional constant value to use as the error if an exception matches (only used if exceptions is not a mapping).	None

Returns:

Type	Description
Callable[[Callable[P_local, Iterator[T_local]]], Callable[P_local, SafeStream[T_local, Any]]]	A decorator that transforms Generator[T] -> SafeStream[T, E].

Examples:

>>> # 1. Simple catch (returns the caught instance)
>>> @catch_each_iter(ValueError)
... def pump(n):
...     for i in range(n):
...         if i == 2:
...             raise ValueError("fail")
...         yield i
>>> list(pump(3))
[Ok(0), Ok(1), Err(ValueError('fail'))]

Source code in src/result/adapters.py

def catch_each_iter[T_local, E_local: Exception, **P_local](
    exceptions: type[E_local] | tuple[type[E_local], ...] | Mapping[type[E_local], Any],
    *,
    map_to: Any = None,
) -> Callable[[Callable[P_local, Iterator[T_local]]], Callable[P_local, SafeStream[T_local, Any]]]:
    """Wrap a generator function to capture iteration-level exceptions into a SafeStream.

    This decorator ensures that if an exception is raised during the iteration
    of the generator, it is caught and yielded as an `Err` variant.

    Args:
        exceptions: One or more exception types to catch, or a mapping of
            exception types to error values.
        map_to: Optional constant value to use as the error if an exception
            matches (only used if `exceptions` is not a mapping).

    Returns:
        A decorator that transforms Generator[T] -> SafeStream[T, E].

    Examples:
        >>> # 1. Simple catch (returns the caught instance)
        >>> @catch_each_iter(ValueError)
        ... def pump(n):
        ...     for i in range(n):
        ...         if i == 2:
        ...             raise ValueError("fail")
        ...         yield i
        >>> list(pump(3))
        [Ok(0), Ok(1), Err(ValueError('fail'))]

    """
    exc_map = _resolve_mapping(exceptions, map_to)  # type: ignore[arg-type]
    catch_tuple = tuple(exc_map.keys())
    has_mapping = map_to is not None or isinstance(exceptions, Mapping)

    def decorator(f: Callable[P_local, Iterator[T_local]]) -> Any:
        @wraps(f)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            __tracebackhide__ = True
            return SafeStream(_wrap_gen_sync(f(*args, **kwargs), catch_tuple, exc_map, has_mapping=has_mapping))

        return wrapper

    return cast("Any", decorator)

catch_each_iter_async

catch_each_iter_async(
    exceptions: type[E_local]
    | tuple[type[E_local], ...]
    | Mapping[type[E_local], Any],
    *,
    map_to: Any = None,
) -> Callable[
    [Callable[P_local, AsyncIterator[T_local]]],
    Callable[P_local, SafeStreamAsync[T_local, Any]],
]

Wrap an async generator function to capture iteration-level exceptions into a SafeStreamAsync.

This is the asynchronous version of @catch_each_iter.

Parameters:

Name	Type	Description	Default
exceptions	type[E_local] | tuple[type[E_local], ...] | Mapping[type[E_local], Any]	One or more exception types to catch, or a mapping of exception types to error values.	required
map_to	Any	Optional constant value to use as the error if an exception matches (only used if exceptions is not a mapping).	None

Returns:

Type	Description
Callable[[Callable[P_local, AsyncIterator[T_local]]], Callable[P_local, SafeStreamAsync[T_local, Any]]]	A decorator that transforms AsyncGenerator[T] -> SafeStreamAsync[T, E].

Source code in src/result/adapters.py

def catch_each_iter_async[T_local, E_local: Exception, **P_local](
    exceptions: type[E_local] | tuple[type[E_local], ...] | Mapping[type[E_local], Any],
    *,
    map_to: Any = None,
) -> Callable[[Callable[P_local, AsyncIterator[T_local]]], Callable[P_local, SafeStreamAsync[T_local, Any]]]:
    """Wrap an async generator function to capture iteration-level exceptions into a SafeStreamAsync.

    This is the asynchronous version of `@catch_each_iter`.

    Args:
        exceptions: One or more exception types to catch, or a mapping of
            exception types to error values.
        map_to: Optional constant value to use as the error if an exception
            matches (only used if `exceptions` is not a mapping).

    Returns:
        A decorator that transforms AsyncGenerator[T] -> SafeStreamAsync[T, E].

    """
    exc_map = _resolve_mapping(exceptions, map_to)  # type: ignore[arg-type]
    catch_tuple = tuple(exc_map.keys())
    has_mapping = map_to is not None or isinstance(exceptions, Mapping)

    def decorator(f: Callable[P_local, AsyncIterator[T_local]]) -> Any:
        @wraps(f)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            __tracebackhide__ = True
            return SafeStreamAsync(_wrap_gen_async(f(*args, **kwargs), catch_tuple, exc_map, has_mapping=has_mapping))

        return wrapper

    return cast("Any", decorator)

catch_instance

catch_instance(
    obj: T_obj,
    exceptions: type[Exception]
    | tuple[type[Exception], ...],
    *,
    map_to: Any = None,
) -> T_obj

catch_instance(
    obj: T_obj, exceptions: Mapping[type[Exception], Any]
) -> T_obj

catch_instance(
    obj: Any, exceptions: Any, *, map_to: Any = None
) -> Any

Wrap a specific object instance so all method calls return Results.

Ideal for third-party objects returned from factories that you don't control the class of.

Parameters:

Name	Type	Description	Default
obj	Any	The instance to wrap.	required
exceptions	Any	The exceptions to catch.	required
map_to	Any	Optional constant error value.	None

Returns:

Type	Description
Any	A proxy object that behaves like the original but wraps methods in @catch.

Static Analysis Note (Type Erasure): Using this proxy causes Type Erasure. Most Python type checkers will believe the returned object is of type T_obj.

Source code in src/result/adapters.py

def catch_instance(
    obj: Any,
    exceptions: Any,
    *,
    map_to: Any = None,
) -> Any:
    """Wrap a specific object instance so all method calls return Results.

    Ideal for third-party objects returned from factories that you don't
    control the class of.

    Args:
        obj: The instance to wrap.
        exceptions: The exceptions to catch.
        map_to: Optional constant error value.

    Returns:
        A proxy object that behaves like the original but wraps methods in @catch.

    Static Analysis Note (Type Erasure):
        Using this proxy causes **Type Erasure**. Most Python type checkers
        will believe the returned object is of type `T_obj`.

    """
    # Cast to Any so the type checker thinks it's the original type
    return cast("Any", _CatchInstanceProxy(obj, exceptions, map_to))

safe_resource

safe_resource(cm: Any) -> _SafeResourceContext[T]

Wrap a standard context manager so that failures in enter yield an Err.

This ensures that the 'with' block is always entered, but the target is a Result variant.

Parameters:

Name	Type	Description	Default
cm	Any	A standard context manager (e.g., from open()).	required

Returns:

Type	Description
_SafeResourceContext[T]	A context manager that yields Result[T, Exception].

Examples:

>>> with safe_resource(open("missing.txt")) as res:
...     match res:
...         case Ok(f):
...             print(f.read())
...         case Err(e):
...             print(f"Error: {e}")

Source code in src/result/adapters.py

def safe_resource[T](cm: Any) -> _SafeResourceContext[T]:
    """Wrap a standard context manager so that failures in __enter__ yield an Err.

    This ensures that the 'with' block is always entered, but the target
    is a Result variant.

    Args:
        cm: A standard context manager (e.g., from open()).

    Returns:
        A context manager that yields Result[T, Exception].

    Examples:
        >>> with safe_resource(open("missing.txt")) as res:
        ...     match res:
        ...         case Ok(f):
        ...             print(f.read())
        ...         case Err(e):
        ...             print(f"Error: {e}")

    """
    return _SafeResourceContext(cm)