From 482a1f9e0fdcd6d8962426505db2deba4c8376fc Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 10:49:39 +0800 Subject: [PATCH 1/8] Add assert_ok utility for Result validation This commit introduces `assert_ok`, a new utility for the `result` library. `assert_ok` can be used as both a standalone function and a context manager to enforce that a `Result` is `Ok`. If an `Err` is encountered, it raises an `AssertionError` with a configurable message. The context manager version also includes automatic scanning of local variables for `Err` assignments, providing fail-fast error detection. This feature enhances the developer experience by providing a concise and explicit way to handle expected successful outcomes and catch unexpected errors early. --- src/result/__init__.py | 3 +- src/result/future.py | 146 ++++++++++++++++++++++++++++++++- tests/result/test_assert_ok.py | 75 +++++++++++++++++ 3 files changed, 219 insertions(+), 5 deletions(-) create mode 100644 tests/result/test_assert_ok.py diff --git a/src/result/__init__.py b/src/result/__init__.py index a201035..ecd817d 100644 --- a/src/result/__init__.py +++ b/src/result/__init__.py @@ -15,7 +15,7 @@ validate, validate_async, ) -from .future import SafeStream, SafeStreamAsync, catch_each_iter, catch_each_iter_async +from .future import SafeStream, SafeStreamAsync, assert_ok, catch_each_iter, catch_each_iter_async from .outcome import Outcome, as_outcome, catch_outcome from .result import ( CatchContext, @@ -58,6 +58,7 @@ "any_ok", "as_err", "as_outcome", + "assert_ok", "catch", "catch_call", "catch_each_iter", diff --git a/src/result/future.py b/src/result/future.py index e7368bc..3e17c3c 100644 --- a/src/result/future.py +++ b/src/result/future.py @@ -11,19 +11,31 @@ """ # pyright: reportPrivateUsage=false +# mypy: disable-error-code="no-any-return" from __future__ import annotations +import sys from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Iterator, Mapping from functools import wraps -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, cast, overload -from .result import Err, Ok, Result, _resolve_mapping, combine, partition +from .result import Err, Ok, OkErr, Result, _resolve_mapping, combine, partition if TYPE_CHECKING: from .outcome import Outcome +def _raise_assertion_error(message: str) -> Any: + """Internal helper to raise AssertionError and hide this frame from traceback.""" + __tracebackhide__ = True + try: + raise AssertionError(message) # noqa: TRY301 + except AssertionError as e: + tb = e.__traceback__ + raise e.with_traceback(tb.tb_next if tb else None) from None + + def _wrap_gen_sync[T, E: Exception]( original_gen: Iterator[T], catch_tuple: tuple[type[E], ...], @@ -120,11 +132,12 @@ def catch_each_iter[T_local, E_local: Exception, **P_local]( 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) # type: ignore[no-any-return] # ty:ignore[unused-type-ignore-comment, unused-ignore-comment] + return cast("Any", decorator) def catch_each_iter_async[T_local, E_local: Exception, **P_local]( @@ -163,11 +176,136 @@ def catch_each_iter_async[T_local, E_local: Exception, **P_local]( 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) # type: ignore[no-any-return] # ty:ignore[unused-type-ignore-comment, unused-ignore-comment] + return cast("Any", decorator) + + +class AssertOk: + """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. + + """ + + 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 + + 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) # noqa: SLF001 + 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 + + 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) + + def _trace_callback(self, frame: Any, event: str, _arg: Any) -> Any: + """Trace function that scans locals for Err variants after each line.""" + __tracebackhide__ = True + # Prevent re-entrancy issues if scanning logic itself triggers tracer + if self._is_scanning: + return self._trace_callback + + if event == "line": + self._is_scanning = True + try: + # Scan current local variables + for name, value in frame.f_locals.items(): + # Only check variables that were added during this block + if name not in self._initial_locals: + match value: + case Err(error_val): # pyright: ignore[reportUnknownVariableType] + # Trigger the error + error_str = cast("Any", error_val) + _raise_assertion_error(f"{self.message}: {error_str}") + case _: + pass + finally: + self._is_scanning = False + return self._trace_callback + + 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 + + +@overload +def assert_ok[T, E](result_or_message: Result[T, E]) -> T: ... + + +@overload +def assert_ok(result_or_message: str = "Result was Err") -> AssertOk: ... + + +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. + + Examples: + >>> # 1. Functional usage + >>> val = assert_ok(Ok(10)) + >>> # assert_ok(Err("fail")) # Raises AssertionError + + >>> # 2. Automatic context manager usage (Fail-fast) + >>> with assert_ok("Critical operations"): + ... res = Ok(1) # Fine + ... # res2 = Err("boom") # Raises AssertionError immediately + + >>> # 3. Explicit check usage + >>> 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) class SafeStream[T, E](Iterable["Result[T, E]"]): diff --git a/tests/result/test_assert_ok.py b/tests/result/test_assert_ok.py new file mode 100644 index 0000000..b2dbb93 --- /dev/null +++ b/tests/result/test_assert_ok.py @@ -0,0 +1,75 @@ +import pytest + +from result import Err, Ok, assert_ok + + +def test_assert_ok_function_success() -> None: + """Verify assert_ok as a function returns value on Ok.""" + expected = 42 + res = Ok(expected) + val = assert_ok(res) + assert val == expected + + +def test_assert_ok_function_failure() -> None: + """Verify assert_ok as a function raises AssertionError on Err.""" + res = Err("something went wrong") + with pytest.raises(AssertionError, match="assert_ok failed: something went wrong"): + assert_ok(res) + + +def test_assert_ok_context_manager_success() -> None: + """Verify assert_ok as a context manager allows multiple checks.""" + v1, v2 = 10, 20 + with assert_ok("Critical operations") as ctx: + val1 = ctx.check(Ok(v1)) + val2 = ctx.check(Ok(v2)) + + assert val1 == v1 + assert val2 == v2 + + +def test_assert_ok_context_manager_failure() -> None: + """Verify assert_ok as a context manager raises AssertionError with custom message.""" + with ( + pytest.raises(AssertionError, match="Database operations: connection refused"), + assert_ok("Database operations") as ctx, + ): + ctx.check(Ok("connected")) + ctx.check(Err("connection refused")) + # This line should never be reached + ctx.check(Ok("done")) + + +def test_assert_ok_nested() -> None: + """Verify assert_ok can be nested.""" + with assert_ok("Outer") as outer: + val_outer = outer.check(Ok("outer_ok")) + with assert_ok("Inner") as inner: + val_inner = inner.check(Ok("inner_ok")) + + assert val_outer == "outer_ok" + assert val_inner == "inner_ok" + + +def test_assert_ok_automatic_scanning() -> None: + """Verify that assert_ok automatically detects Err assignments.""" + with ( + pytest.raises(AssertionError, match="Auto-check: boom"), + assert_ok("Auto-check"), + ): + # This assignment should be caught by the tracer + res = Err("boom") + # This line should never be reached + print(res) + + +def test_assert_ok_automatic_scanning_ignore_existing() -> None: + """Verify that assert_ok ignores Err variables that existed before the block.""" + existing_err = Err("already here") + + with assert_ok("Should pass"): + # Accessing an existing Err shouldn't trigger if it's not reassigned + _ = existing_err + x = Ok(1) + assert x.is_ok() From 44c67643f7285facb9c7a98ecbb8e5b66e717931 Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 10:54:28 +0800 Subject: [PATCH 2/8] Refactor assert_ok docstring for clarity --- src/result/future.py | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/result/future.py b/src/result/future.py index 3e17c3c..c4059f6 100644 --- a/src/result/future.py +++ b/src/result/future.py @@ -280,17 +280,35 @@ def assert_ok(result_or_message: Any = "Result was Err") -> Any: 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 + >>> # 1. Functional usage (Low overhead) >>> val = assert_ok(Ok(10)) >>> # assert_ok(Err("fail")) # Raises AssertionError - >>> # 2. Automatic context manager usage (Fail-fast) + >>> # 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 + >>> # 3. Explicit check usage (Lower overhead in context) >>> with assert_ok() as ctx: ... val = ctx.check(Ok(42)) From 90aaec89d7c89347d620f70c6362ab1f9bfef6b0 Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 11:02:11 +0800 Subject: [PATCH 3/8] feat: Add catch_boundary and catch_instance utilities Introduce `catch_boundary` for class-level exception wrapping and `catch_instance` for wrapping individual object instances. These utilities simplify the integration of external libraries or classes that raise exceptions into the Result type system. --- src/result/__init__.py | 12 ++++- src/result/future.py | 102 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/src/result/__init__.py b/src/result/__init__.py index ecd817d..ea3b19b 100644 --- a/src/result/__init__.py +++ b/src/result/__init__.py @@ -15,7 +15,15 @@ validate, validate_async, ) -from .future import SafeStream, SafeStreamAsync, assert_ok, catch_each_iter, catch_each_iter_async +from .future import ( + SafeStream, + SafeStreamAsync, + assert_ok, + catch_boundary, + catch_each_iter, + catch_each_iter_async, + catch_instance, +) from .outcome import Outcome, as_outcome, catch_outcome from .result import ( CatchContext, @@ -60,9 +68,11 @@ "as_outcome", "assert_ok", "catch", + "catch_boundary", "catch_call", "catch_each_iter", "catch_each_iter_async", + "catch_instance", "catch_outcome", "combine", "do", diff --git a/src/result/future.py b/src/result/future.py index c4059f6..64a402f 100644 --- a/src/result/future.py +++ b/src/result/future.py @@ -15,6 +15,7 @@ from __future__ import annotations +import inspect import sys from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Iterator, Mapping from functools import wraps @@ -184,6 +185,107 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: return cast("Any", decorator) +def catch_boundary( + exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + *, + map_to: Any = None, +) -> Callable[[type[Any]], type[Any]]: + """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. + + Examples: + >>> @catch_boundary(ValueError, map_to="domain_error") + ... class Client: + ... def perform(self, x): + ... if x < 0: raise ValueError + ... return x + >>> Client().perform(-1) + Err('domain_error') + + """ + from .result import catch + + exc_map = _resolve_mapping(exceptions, map_to) + + def decorator(cls: type[Any]) -> type[Any]: + for name, method in inspect.getmembers(cls, predicate=inspect.isroutine): + if name.startswith("_"): + continue + # Wrap the method with @catch using the pre-resolved mapping + setattr(cls, name, catch(exc_map)(method)) + return cls + + return decorator + + +class _CatchInstanceProxy: + """Internal proxy that wraps all method calls of an instance with @catch.""" + + def __init__(self, obj: Any, exc_map: dict[type[Exception], Any]) -> None: + # Use object.__setattr__ to avoid infinite recursion with __getattr__ + object.__setattr__(self, "_obj", obj) + object.__setattr__(self, "_exc_map", exc_map) + + def __getattr__(self, name: str) -> Any: + from .result import catch + + obj = object.__getattribute__(self, "_obj") + attr = getattr(obj, name) + exc_map = object.__getattribute__(self, "_exc_map") + + if inspect.isroutine(attr): + # Bind the routine to the original object to ensure 'self' is passed + # This is important for instance methods + bound_method = attr.__get__(obj, obj.__class__) + return catch(exc_map)(bound_method) + return attr + + def __repr__(self) -> str: + obj = object.__getattribute__(self, "_obj") + return f"catch_instance({obj!r})" + + +def catch_instance[T_obj]( + obj: T_obj, + exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + *, + map_to: Any = None, +) -> T_obj: + """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. + + Examples: + >>> class Raw: + ... def run(self): raise ValueError("fail") + >>> safe = catch_instance(Raw(), ValueError) + >>> safe.run() + Err(ValueError('fail')) + + """ + exc_map = _resolve_mapping(exceptions, map_to) + # Cast to T_obj so the type checker thinks it's the original type + return cast("T_obj", _CatchInstanceProxy(obj, exc_map)) + + class AssertOk: """A context manager for asserting that Results must be Ok. From 80f401edb13751286f40ad00c28b88bfc16373b6 Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 11:21:29 +0800 Subject: [PATCH 4/8] Refactor: Add `catch_boundary` and `catch_instance` --- src/result/future.py | 102 ++++++++++++++++++++------ tests/result/test_catch_adapters.py | 110 ++++++++++++++++++++++++++++ 2 files changed, 188 insertions(+), 24 deletions(-) create mode 100644 tests/result/test_catch_adapters.py diff --git a/src/result/future.py b/src/result/future.py index 64a402f..500c790 100644 --- a/src/result/future.py +++ b/src/result/future.py @@ -11,7 +11,7 @@ """ # pyright: reportPrivateUsage=false -# mypy: disable-error-code="no-any-return" +# mypy: disable-error-code="no-any-return, redundant-cast" from __future__ import annotations @@ -19,13 +19,16 @@ import sys from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Iterator, Mapping from functools import wraps -from typing import TYPE_CHECKING, Any, cast, overload +from typing import TYPE_CHECKING, Any, TypeVar, cast, overload -from .result import Err, Ok, OkErr, Result, _resolve_mapping, combine, partition +from .result import Err, Ok, OkErr, Result, _resolve_mapping, catch, combine, partition if TYPE_CHECKING: from .outcome import Outcome +T_cls = TypeVar("T_cls", bound=type[Any]) +T_obj = TypeVar("T_obj") + def _raise_assertion_error(message: str) -> Any: """Internal helper to raise AssertionError and hide this frame from traceback.""" @@ -185,11 +188,25 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: return cast("Any", decorator) +@overload +def catch_boundary( + exceptions: type[Exception] | tuple[type[Exception], ...], + *, + map_to: Any = None, +) -> Callable[[T_cls], T_cls]: ... + + +@overload +def catch_boundary( + exceptions: Mapping[type[Exception], Any], +) -> Callable[[T_cls], T_cls]: ... + + def catch_boundary( - exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + exceptions: Any, *, map_to: Any = None, -) -> Callable[[type[Any]], type[Any]]: +) -> Any: """Wrap all public methods of a class with the @catch decorator. This is an 'Entry Adapter' that allows lifting an entire external SDK or @@ -202,26 +219,32 @@ def catch_boundary( 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]`. + - Your IDE may still show the original return types. + - You may need to use `Any` or explicit type stubs when calling + decorated methods to avoid false-positive type errors. + Examples: >>> @catch_boundary(ValueError, map_to="domain_error") ... class Client: ... def perform(self, x): - ... if x < 0: raise ValueError + ... if x < 0: + ... raise ValueError ... return x >>> Client().perform(-1) Err('domain_error') """ - from .result import catch - - exc_map = _resolve_mapping(exceptions, map_to) - def decorator(cls: type[Any]) -> type[Any]: + 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 the pre-resolved mapping - setattr(cls, name, catch(exc_map)(method)) + # Wrap the method with @catch using original parameters + setattr(cls, name, catch(exceptions, map_to=map_to)(method)) return cls return decorator @@ -230,23 +253,31 @@ def decorator(cls: type[Any]) -> type[Any]: class _CatchInstanceProxy: """Internal proxy that wraps all method calls of an instance with @catch.""" - def __init__(self, obj: Any, exc_map: dict[type[Exception], Any]) -> None: + def __init__( + self, + obj: Any, + exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + map_to: Any = None, + ) -> None: # Use object.__setattr__ to avoid infinite recursion with __getattr__ object.__setattr__(self, "_obj", obj) - object.__setattr__(self, "_exc_map", exc_map) + object.__setattr__(self, "_exceptions", exceptions) + object.__setattr__(self, "_map_to", map_to) def __getattr__(self, name: str) -> Any: - from .result import catch - obj = object.__getattribute__(self, "_obj") attr = getattr(obj, name) - exc_map = object.__getattribute__(self, "_exc_map") + exceptions = object.__getattribute__(self, "_exceptions") + map_to = object.__getattribute__(self, "_map_to") + + if name.startswith("_"): + return attr if inspect.isroutine(attr): # Bind the routine to the original object to ensure 'self' is passed # This is important for instance methods - bound_method = attr.__get__(obj, obj.__class__) - return catch(exc_map)(bound_method) + bound_method = attr.__get__(obj, obj.__class__) if hasattr(attr, "__get__") else attr # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType, reportAttributeAccessIssue] + return catch(exceptions, map_to=map_to)(bound_method) # pyright: ignore[reportUnknownArgumentType] return attr def __repr__(self) -> str: @@ -254,12 +285,28 @@ def __repr__(self) -> str: return f"catch_instance({obj!r})" +@overload def catch_instance[T_obj]( obj: T_obj, - exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + exceptions: type[Exception] | tuple[type[Exception], ...], *, map_to: Any = None, -) -> T_obj: +) -> T_obj: ... + + +@overload +def catch_instance[T_obj]( + obj: T_obj, + exceptions: Mapping[type[Exception], Any], +) -> T_obj: ... + + +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 @@ -273,17 +320,24 @@ def catch_instance[T_obj]( 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` (with original + return types), but at runtime every method will return a `Result`. + - You may need to cast the result to `Any` or use `# type: ignore` + when calling methods on the proxy to satisfy the type checker. + Examples: >>> class Raw: - ... def run(self): raise ValueError("fail") + ... def run(self): + ... raise ValueError("fail") >>> safe = catch_instance(Raw(), ValueError) >>> safe.run() Err(ValueError('fail')) """ - exc_map = _resolve_mapping(exceptions, map_to) # Cast to T_obj so the type checker thinks it's the original type - return cast("T_obj", _CatchInstanceProxy(obj, exc_map)) + return cast("Any", _CatchInstanceProxy(obj, exceptions, map_to)) class AssertOk: diff --git a/tests/result/test_catch_adapters.py b/tests/result/test_catch_adapters.py new file mode 100644 index 0000000..e0a7e14 --- /dev/null +++ b/tests/result/test_catch_adapters.py @@ -0,0 +1,110 @@ +import asyncio +from typing import Any + +import pytest + +from result import Err, Ok, catch_boundary, catch_instance, is_err + +# --- Classes for testing --- + + +class RawClient: + def sync_ok(self, x: int) -> int: + return x + + def sync_fail(self) -> None: + raise ValueError("sync fail") + + async def async_ok(self, x: int) -> int: + await asyncio.sleep(0) + return x + + async def async_fail(self) -> None: + await asyncio.sleep(0) + raise ValueError("async fail") + + def _private(self) -> str: + return "private" + + +@catch_boundary(ValueError) +class SafeClient: + def ok(self, x: int) -> Any: + return x + + def fail(self) -> Any: + raise ValueError("boundary fail") + + async def aok(self, x: int) -> Any: + return x + + async def afail(self) -> Any: + raise ValueError("async boundary fail") + + +# --- Tests --- + + +def test_catch_boundary_sync() -> None: + client: Any = SafeClient() + assert client.ok(10) == Ok(10) + res: Any = client.fail() + assert is_err(res) + # Check that it's a ValueError with the right message + err: Any = res.err() + assert isinstance(err, ValueError) + assert str(err) == "boundary fail" + + +@pytest.mark.asyncio +async def test_catch_boundary_async() -> None: + client: Any = SafeClient() + assert await client.aok(20) == Ok(20) + res: Any = await client.afail() + assert is_err(res) + err: Any = res.err() + assert isinstance(err, ValueError) + assert str(err) == "async boundary fail" + + +def test_catch_boundary_mapping() -> None: + @catch_boundary(ValueError, map_to="mapped") + class MappedClient: + def fail(self) -> Any: + raise ValueError + + assert MappedClient().fail() == Err("mapped") + + +def test_catch_instance_sync() -> None: + raw = RawClient() + safe: Any = catch_instance(raw, ValueError) + + assert safe.sync_ok(42) == Ok(42) + res: Any = safe.sync_fail() + assert is_err(res) # ty: ignore[invalid-argument-type] + err: Any = res.err() # ty: ignore[unresolved-attribute] + assert isinstance(err, ValueError) + assert str(err) == "sync fail" + + # Verify private methods are not wrapped + assert safe._private() == "private" # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_catch_instance_async() -> None: + raw = RawClient() + safe: Any = catch_instance(raw, ValueError) + + assert await safe.async_ok(100) == Ok(100) + res: Any = await safe.async_fail() + assert is_err(res) # ty: ignore[invalid-argument-type] + err: Any = res.err() # ty: ignore[unresolved-attribute] + assert isinstance(err, ValueError) + assert str(err) == "async fail" + + +def test_catch_instance_repr() -> None: + raw = RawClient() + safe = catch_instance(raw, ValueError) + assert "catch_instance" in repr(safe) From a44193eb4585030517210ef6b93dfb2b4c1f36af Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 12:16:47 +0800 Subject: [PATCH 5/8] feat: Add retry and retry_async decorators Introduces `retry_result` and `retry_result_async` decorators to the `future` module, allowing for configurable retries of operations that return a `Result`. These decorators support specifying the number of attempts, a `retry_if` predicate for selective retries, and a `catch` parameter to handle exceptions by converting them into `Err` variants. Additionally, they include support for backoff delays. --- src/result/__init__.py | 4 + src/result/future.py | 363 +++++++++++++++++++++++++++++++++++-- tests/result/test_retry.py | 176 ++++++++++++++++++ 3 files changed, 524 insertions(+), 19 deletions(-) create mode 100644 tests/result/test_retry.py diff --git a/src/result/__init__.py b/src/result/__init__.py index ea3b19b..6e79810 100644 --- a/src/result/__init__.py +++ b/src/result/__init__.py @@ -23,6 +23,8 @@ catch_each_iter, catch_each_iter_async, catch_instance, + retry_result, + retry_result_async, ) from .outcome import Outcome, as_outcome, catch_outcome from .result import ( @@ -91,6 +93,8 @@ "partition_exceptions", "partition_map", "partition_results", + "retry_result", + "retry_result_async", "succeeds", "traverse", "traverse_async", diff --git a/src/result/future.py b/src/result/future.py index 500c790..7d2a5f1 100644 --- a/src/result/future.py +++ b/src/result/future.py @@ -15,13 +15,17 @@ from __future__ import annotations +import asyncio import inspect +import random import sys -from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Iterator, Mapping +import time +from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Iterable, Iterator, Mapping from functools import wraps from typing import TYPE_CHECKING, Any, TypeVar, cast, overload from .result import Err, Ok, OkErr, Result, _resolve_mapping, catch, combine, partition +from .result import catch as _catch_decorator if TYPE_CHECKING: from .outcome import Outcome @@ -188,25 +192,11 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: return cast("Any", decorator) -@overload def catch_boundary( - exceptions: type[Exception] | tuple[type[Exception], ...], - *, - map_to: Any = None, -) -> Callable[[T_cls], T_cls]: ... - - -@overload -def catch_boundary( - exceptions: Mapping[type[Exception], Any], -) -> Callable[[T_cls], T_cls]: ... - - -def catch_boundary( - exceptions: Any, + exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], *, map_to: Any = None, -) -> Any: +) -> 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 @@ -244,7 +234,8 @@ def decorator(cls: T_cls) -> T_cls: if name.startswith("_"): continue # Wrap the method with @catch using original parameters - setattr(cls, name, catch(exceptions, map_to=map_to)(method)) + # Use Any to satisfy ty's overload resolution + setattr(cls, name, catch(cast("Any", exceptions), map_to=map_to)(method)) return cls return decorator @@ -336,7 +327,7 @@ def catch_instance( Err(ValueError('fail')) """ - # Cast to T_obj so the type checker thinks it's the original type + # Cast to Any so the type checker thinks it's the original type return cast("Any", _CatchInstanceProxy(obj, exceptions, map_to)) @@ -619,3 +610,337 @@ async def to_outcome(self) -> Outcome[list[T], list[E]]: ) from None return Outcome(oks, errs or None) + + +def _get_retry_delay( + current_delay: float, + *, + jitter: bool | float, +) -> float: + """Internal helper to calculate next retry delay.""" + if current_delay <= 0: + return 0 + sleep_time = current_delay + if jitter: + # If jitter is True, default to 0.1, else use the provided float value + jitter_val = jitter if isinstance(jitter, float) else 0.1 + sleep_time += random.uniform(0, jitter_val) + return sleep_time + + +@overload +def retry_result[T, E, **P]( + 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]]]: ... + + +@overload +def retry_result[T, E, **P]( + 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]]]: ... + + +@overload +def retry_result[T, E_exc: Exception, **P]( + 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]]]: ... + + +@overload +def retry_result[T, **P]( + 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]]]: ... + + +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]]: + # If catch is provided, we wrap the function in @catch first + wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) + + +@overload +def retry_result_async[T, E, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, E, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, E_exc: Exception, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, **P]( + 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]]]]: ... + + +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 + wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) diff --git a/tests/result/test_retry.py b/tests/result/test_retry.py new file mode 100644 index 0000000..c826191 --- /dev/null +++ b/tests/result/test_retry.py @@ -0,0 +1,176 @@ +import asyncio +from typing import Any +from unittest.mock import patch + +import pytest + +from result import Err, Ok, Result, catch, retry_result, retry_result_async + + +def test_retry_sync_success() -> None: + calls = 0 + attempts = 3 + + @retry_result(attempts=attempts) + def op() -> Ok[int] | Err[str]: + nonlocal calls + calls += 1 + expected_calls_for_success = 2 + if calls < expected_calls_for_success: + return Err("fail") + return Ok(42) + + res: Result[int, str] = op() + assert res == Ok(42) + expected_total_calls = 2 + assert calls == expected_total_calls + + +def test_retry_sync_exhausted() -> None: + calls = 0 + attempts = 3 + + @retry_result(attempts=attempts) + def op() -> Err[str]: + nonlocal calls + calls += 1 + return Err("fail") + + res: Result[Any, str] = op() # pyright: ignore[reportUnknownVariableType] + assert res == Err("fail") + assert calls == attempts + + +def test_retry_if_predicate() -> None: + calls = 0 + attempts_limit = 5 + + @retry_result(attempts=attempts_limit, retry_if=lambda e: e == "retryable") + def op() -> Err[str]: + nonlocal calls + calls += 1 + first_call = 1 + if calls == first_call: + return Err("retryable") + return Err("fatal") + + res: Result[Any, str] = op() # pyright: ignore[reportUnknownVariableType] + assert res == Err("fatal") + expected_calls = 2 + assert calls == expected_calls + + +def test_retry_with_catch_sync() -> None: + calls = 0 + attempts = 3 + + @retry_result(attempts=attempts, catch=ValueError) + def op() -> int: + nonlocal calls + calls += 1 + expected_calls_for_success = 3 + if calls < expected_calls_for_success: + raise ValueError("fail") + return 42 + + res: Result[int, Any] = op() # pyright: ignore[reportUnknownVariableType] + assert res == Ok(42) + assert calls == attempts + + +def test_retry_with_catch_exhausted() -> None: + calls = 0 + attempts = 3 + + @retry_result(attempts=attempts, catch=ValueError) + def op() -> int: + nonlocal calls + calls += 1 + raise ValueError("fail") + + res: Result[int, Any] = op() # pyright: ignore[reportUnknownVariableType] + assert isinstance(res, Err) + err = res.err() + assert isinstance(err, ValueError) + assert calls == attempts + + +def test_retry_plus_catch_stacking() -> None: + # Verify that putting retry on TOP of catch works without the 'catch' parameter + calls = 0 + attempts = 3 + + @retry_result(attempts=attempts) + @catch(ValueError) + def op() -> int: + nonlocal calls + calls += 1 + expected_calls_for_success = 2 + if calls < expected_calls_for_success: + raise ValueError("fail") + return 42 + + res = op() + assert res == Ok(42) + expected_total_calls = 2 + assert calls == expected_total_calls + + +@pytest.mark.asyncio +async def test_retry_async_success() -> None: + calls = 0 + attempts = 3 + + @retry_result_async(attempts=attempts, delay=0.01) + async def op() -> Ok[int] | Err[str]: + nonlocal calls + calls += 1 + # Use sleep to satisfy async + await asyncio.sleep(0) + expected_calls_for_success = 3 + if calls < expected_calls_for_success: + return Err("transient") + return Ok(100) + + res: Result[int, str] = await op() # pyright: ignore[reportUnknownVariableType] + assert res == Ok(100) + assert calls == attempts + + +@pytest.mark.asyncio +async def test_retry_with_catch_async() -> None: + calls = 0 + attempts = 3 + + @retry_result_async(attempts=attempts, catch=ValueError) + async def op() -> int: + nonlocal calls + calls += 1 + # use await to avoid RUF029 + await asyncio.sleep(0) + expected_calls_for_success = 2 + if calls < expected_calls_for_success: + raise ValueError("async fail") + return 100 + + res: Result[int, Any] = await op() # pyright: ignore[reportUnknownVariableType] + assert res == Ok(100) + expected_total_calls = 2 + assert calls == expected_total_calls + + +def test_retry_backoff_logic() -> None: + with patch("time.sleep") as mock_sleep: + attempts = 3 + initial_delay = 0.1 + backoff_factor = 2.0 + + @retry_result(attempts=attempts, delay=initial_delay, backoff=backoff_factor) + def op() -> Err[str]: + return Err("fail") + + op() + expected_sleep_count = 2 + assert mock_sleep.call_count == expected_sleep_count + mock_sleep.assert_any_call(0.1) + mock_sleep.assert_any_call(0.2) From 2458e15142c580c43f31068d2170e35868689933 Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 13:02:36 +0800 Subject: [PATCH 6/8] Refactor: Move adapter utilities from future to result The `future.py` module contained adapter utilities like `SafeStream`, `SafeStreamAsync`, `catch_boundary`, `catch_each_iter`, `catch_each_iter_async`, and `catch_instance`. These utilities are now core functionality and have been moved to the `result.adapters` submodule. This refactoring: - Consolidates related functionality into a dedicated module. - Removes the `future.py` module, which was marked as experimental and may have caused confusion. - Updates imports in `result/__init__.py` to reflect the new module structure. - Updates the documentation `docs/api/adapters.md` to point to the correct module. --- docs/api/adapters.md | 5 + src/result/__init__.py | 22 +- src/result/adapters.py | 356 +++++++++++ src/result/future.py | 946 ---------------------------- src/result/result.py | 498 ++++++++++++++- tests/result/test_catch_adapters.py | 8 +- 6 files changed, 872 insertions(+), 963 deletions(-) create mode 100644 docs/api/adapters.md create mode 100644 src/result/adapters.py delete mode 100644 src/result/future.py diff --git a/docs/api/adapters.md b/docs/api/adapters.md new file mode 100644 index 0000000..1c08554 --- /dev/null +++ b/docs/api/adapters.md @@ -0,0 +1,5 @@ +# Adapters + +Bulk lifting and integration utilities for the Result Pattern. + +::: result.adapters diff --git a/src/result/__init__.py b/src/result/__init__.py index 6e79810..bb7ca9b 100644 --- a/src/result/__init__.py +++ b/src/result/__init__.py @@ -1,3 +1,11 @@ +from .adapters import ( + SafeStream, + SafeStreamAsync, + catch_boundary, + catch_each_iter, + catch_each_iter_async, + catch_instance, +) from .combinators import ( add_context, ensure, @@ -15,17 +23,6 @@ validate, validate_async, ) -from .future import ( - SafeStream, - SafeStreamAsync, - assert_ok, - catch_boundary, - catch_each_iter, - catch_each_iter_async, - catch_instance, - retry_result, - retry_result_async, -) from .outcome import Outcome, as_outcome, catch_outcome from .result import ( CatchContext, @@ -38,6 +35,7 @@ UnwrapError, any_ok, as_err, + assert_ok, catch, catch_call, combine, @@ -50,6 +48,8 @@ is_ok, map2, partition, + retry_result, + retry_result_async, ) __all__ = [ diff --git a/src/result/adapters.py b/src/result/adapters.py new file mode 100644 index 0000000..02df3f9 --- /dev/null +++ b/src/result/adapters.py @@ -0,0 +1,356 @@ +"""# 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. + +""" + +# pyright: reportPrivateUsage=false +# mypy: disable-error-code="no-any-return, redundant-cast" + +from __future__ import annotations + +import inspect +from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Iterator, Mapping +from functools import wraps +from typing import TYPE_CHECKING, Any, TypeVar, cast, overload + +from .result import Err, Ok, Result, _resolve_mapping, catch, combine, partition + +if TYPE_CHECKING: + from .outcome import Outcome + +T_cls = TypeVar("T_cls", bound=type[Any]) +T_obj = TypeVar("T_obj") + + +def _wrap_gen_sync[T, E: Exception]( + original_gen: Iterator[T], + catch_tuple: tuple[type[E], ...], + exc_map: Mapping[type[Exception], Any], + *, + has_mapping: bool, +) -> Iterator[Result[T, Any]]: + """Internal helper to wrap a synchronous iterator with exception handling.""" + try: + for val in original_gen: + yield Ok(val) + except catch_tuple as e: + yield Err(exc_map.get(type(e), e) if has_mapping else e) + except Exception as e: # noqa: BLE001 + tb = e.__traceback__ + raise e.with_traceback(tb.tb_next if tb else None) from None + + +async def _wrap_gen_async[T, E: Exception]( + original_gen: AsyncIterator[T], + catch_tuple: tuple[type[E], ...], + exc_map: Mapping[type[Exception], Any], + *, + has_mapping: bool, +) -> AsyncIterator[Result[T, Any]]: + """Internal helper to wrap an asynchronous iterator with exception handling.""" + try: + async for val in original_gen: + yield Ok(val) + except catch_tuple as e: + yield Err(exc_map.get(type(e), e) if has_mapping else e) + except Exception as e: # noqa: BLE001 + tb = e.__traceback__ + raise e.with_traceback(tb.tb_next if tb else None) from None + + +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) + + +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) + + +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 + + +class _CatchInstanceProxy: + """Internal proxy that wraps all method calls of an instance with @catch.""" + + def __init__( + self, + obj: Any, + exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], + map_to: Any = None, + ) -> None: + # Use object.__setattr__ to avoid infinite recursion with __getattr__ + object.__setattr__(self, "_obj", obj) + object.__setattr__(self, "_exceptions", exceptions) + object.__setattr__(self, "_map_to", map_to) + + def __getattr__(self, name: str) -> Any: + obj = object.__getattribute__(self, "_obj") + attr = getattr(obj, name) + exceptions = object.__getattribute__(self, "_exceptions") + map_to = object.__getattribute__(self, "_map_to") + + if name.startswith("_"): + return attr + + if inspect.isroutine(attr): + # Bind the routine to the original object to ensure 'self' is passed + # This is important for instance methods + bound_method = attr.__get__(obj, obj.__class__) if hasattr(attr, "__get__") else attr # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType, reportAttributeAccessIssue] + return catch(exceptions, map_to=map_to)(bound_method) # pyright: ignore[reportUnknownArgumentType] + return None + + def __repr__(self) -> str: + obj = object.__getattribute__(self, "_obj") + return f"catch_instance({obj!r})" + + +@overload +def catch_instance[T_obj]( + obj: T_obj, + exceptions: type[Exception] | tuple[type[Exception], ...], + *, + map_to: Any = None, +) -> T_obj: ... + + +@overload +def catch_instance[T_obj]( + obj: T_obj, + exceptions: Mapping[type[Exception], Any], +) -> T_obj: ... + + +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)) + + +class SafeStream[T, E](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. + + """ + + def __init__(self, gen: Iterator[Result[T, E]]) -> None: + """Initialize a SafeStream with a Result-yielding iterator.""" + self._gen = gen + self._consumed = False + + 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 + + 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)) + + 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) + + +class SafeStreamAsync[T, E](AsyncIterable["Result[T, E]"]): + """Async version of SafeStream for fallible asynchronous generators.""" + + def __init__(self, gen: AsyncIterator[Result[T, E]]) -> None: + """Initialize a SafeStreamAsync with a Result-yielding async iterator.""" + self._gen = gen + self._consumed = False + + 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 + + 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) + + 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) diff --git a/src/result/future.py b/src/result/future.py deleted file mode 100644 index 7d2a5f1..0000000 --- a/src/result/future.py +++ /dev/null @@ -1,946 +0,0 @@ -"""# Future: Experimental and Alpha features for Result Pattern. - -This module houses features that are currently in testing or alpha stage. -These features are designed to handle iteration-level errors and provide -fault-tolerant streaming primitives. - -Note: - API stability is not guaranteed. These features may change or be removed - in future versions without a major version bump. - -""" - -# pyright: reportPrivateUsage=false -# mypy: disable-error-code="no-any-return, redundant-cast" - -from __future__ import annotations - -import asyncio -import inspect -import random -import sys -import time -from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Iterable, Iterator, Mapping -from functools import wraps -from typing import TYPE_CHECKING, Any, TypeVar, cast, overload - -from .result import Err, Ok, OkErr, Result, _resolve_mapping, catch, combine, partition -from .result import catch as _catch_decorator - -if TYPE_CHECKING: - from .outcome import Outcome - -T_cls = TypeVar("T_cls", bound=type[Any]) -T_obj = TypeVar("T_obj") - - -def _raise_assertion_error(message: str) -> Any: - """Internal helper to raise AssertionError and hide this frame from traceback.""" - __tracebackhide__ = True - try: - raise AssertionError(message) # noqa: TRY301 - except AssertionError as e: - tb = e.__traceback__ - raise e.with_traceback(tb.tb_next if tb else None) from None - - -def _wrap_gen_sync[T, E: Exception]( - original_gen: Iterator[T], - catch_tuple: tuple[type[E], ...], - exc_map: Mapping[type[Exception], Any], - *, - has_mapping: bool, -) -> Iterator[Result[T, Any]]: - """Internal helper to wrap a synchronous iterator with exception handling.""" - try: - for val in original_gen: - yield Ok(val) - except catch_tuple as e: - yield Err(exc_map.get(type(e), e) if has_mapping else e) - except Exception as e: # noqa: BLE001 - tb = e.__traceback__ - raise e.with_traceback(tb.tb_next if tb else None) from None - - -async def _wrap_gen_async[T, E: Exception]( - original_gen: AsyncIterator[T], - catch_tuple: tuple[type[E], ...], - exc_map: Mapping[type[Exception], Any], - *, - has_mapping: bool, -) -> AsyncIterator[Result[T, Any]]: - """Internal helper to wrap an asynchronous iterator with exception handling.""" - try: - async for val in original_gen: - yield Ok(val) - except catch_tuple as e: - yield Err(exc_map.get(type(e), e) if has_mapping else e) - except Exception as e: # noqa: BLE001 - tb = e.__traceback__ - raise e.with_traceback(tb.tb_next if tb else None) from None - - -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'))] - - >>> # 2. Catch with map_to - >>> @catch_each_iter(ValueError, map_to="error") - ... def pump_mapped(n): - ... for i in range(n): - ... if i == 1: - ... raise ValueError - ... yield i - >>> list(pump_mapped(2)) - [Ok(0), Err('error')] - - >>> # 3. Catch multiple with mapping - >>> err_map = {ValueError: "val_err", TypeError: "type_err"} - >>> @catch_each_iter(err_map) - ... def pump_multi(x): - ... if x == 0: - ... raise ValueError - ... if x == 1: - ... raise TypeError - ... yield "ok" - >>> list(pump_multi(0)) - [Err('val_err')] - - """ - 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) - - -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]. - - Examples: - >>> @catch_each_iter_async(ValueError) - ... async def async_pump(n): - ... for i in range(n): - ... if i == 1: - ... raise ValueError("async fail") - ... yield i - >>> [res async for res in async_pump(2)] - [Ok(0), Err(ValueError('async 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, 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) - - -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]`. - - Your IDE may still show the original return types. - - You may need to use `Any` or explicit type stubs when calling - decorated methods to avoid false-positive type errors. - - Examples: - >>> @catch_boundary(ValueError, map_to="domain_error") - ... class Client: - ... def perform(self, x): - ... if x < 0: - ... raise ValueError - ... return x - >>> Client().perform(-1) - Err('domain_error') - - """ - - 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 - - -class _CatchInstanceProxy: - """Internal proxy that wraps all method calls of an instance with @catch.""" - - def __init__( - self, - obj: Any, - exceptions: type[Exception] | tuple[type[Exception], ...] | Mapping[type[Exception], Any], - map_to: Any = None, - ) -> None: - # Use object.__setattr__ to avoid infinite recursion with __getattr__ - object.__setattr__(self, "_obj", obj) - object.__setattr__(self, "_exceptions", exceptions) - object.__setattr__(self, "_map_to", map_to) - - def __getattr__(self, name: str) -> Any: - obj = object.__getattribute__(self, "_obj") - attr = getattr(obj, name) - exceptions = object.__getattribute__(self, "_exceptions") - map_to = object.__getattribute__(self, "_map_to") - - if name.startswith("_"): - return attr - - if inspect.isroutine(attr): - # Bind the routine to the original object to ensure 'self' is passed - # This is important for instance methods - bound_method = attr.__get__(obj, obj.__class__) if hasattr(attr, "__get__") else attr # pyright: ignore[reportUnknownVariableType, reportUnknownMemberType, reportAttributeAccessIssue] - return catch(exceptions, map_to=map_to)(bound_method) # pyright: ignore[reportUnknownArgumentType] - return attr - - def __repr__(self) -> str: - obj = object.__getattribute__(self, "_obj") - return f"catch_instance({obj!r})" - - -@overload -def catch_instance[T_obj]( - obj: T_obj, - exceptions: type[Exception] | tuple[type[Exception], ...], - *, - map_to: Any = None, -) -> T_obj: ... - - -@overload -def catch_instance[T_obj]( - obj: T_obj, - exceptions: Mapping[type[Exception], Any], -) -> T_obj: ... - - -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` (with original - return types), but at runtime every method will return a `Result`. - - You may need to cast the result to `Any` or use `# type: ignore` - when calling methods on the proxy to satisfy the type checker. - - Examples: - >>> class Raw: - ... def run(self): - ... raise ValueError("fail") - >>> safe = catch_instance(Raw(), ValueError) - >>> safe.run() - Err(ValueError('fail')) - - """ - # Cast to Any so the type checker thinks it's the original type - return cast("Any", _CatchInstanceProxy(obj, exceptions, map_to)) - - -class AssertOk: - """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. - - """ - - 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 - - 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) # noqa: SLF001 - 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 - - 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) - - def _trace_callback(self, frame: Any, event: str, _arg: Any) -> Any: - """Trace function that scans locals for Err variants after each line.""" - __tracebackhide__ = True - # Prevent re-entrancy issues if scanning logic itself triggers tracer - if self._is_scanning: - return self._trace_callback - - if event == "line": - self._is_scanning = True - try: - # Scan current local variables - for name, value in frame.f_locals.items(): - # Only check variables that were added during this block - if name not in self._initial_locals: - match value: - case Err(error_val): # pyright: ignore[reportUnknownVariableType] - # Trigger the error - error_str = cast("Any", error_val) - _raise_assertion_error(f"{self.message}: {error_str}") - case _: - pass - finally: - self._is_scanning = False - return self._trace_callback - - 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 - - -@overload -def assert_ok[T, E](result_or_message: Result[T, E]) -> T: ... - - -@overload -def assert_ok(result_or_message: str = "Result was Err") -> AssertOk: ... - - -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) - - -class SafeStream[T, E](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. - - Attributes: - _gen: The internal iterator yielding Results. - _consumed: Whether the stream has already been iterated. - - """ - - def __init__(self, gen: Iterator[Result[T, E]]) -> None: - """Initialize a SafeStream with a Result-yielding iterator.""" - self._gen = gen - self._consumed = False - - 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 - - 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. - - Returns: - The combined Result of the stream. - - Examples: - >>> @catch_each_iter(ValueError) - ... def gen(fail): - ... yield 1 - ... if fail: - ... raise ValueError("fail") - ... yield 2 - >>> gen(fail=False).to_result() - Ok([1, 2]) - >>> gen(fail=True).to_result() - Err(ValueError('fail')) - - """ - return combine(list(self)) - - 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. - - Returns: - An Outcome containing two lists: successes and errors. - - Examples: - >>> @catch_each_iter(ValueError) - ... def gen(): - ... yield 1 - ... raise ValueError("err") - >>> out = gen().to_outcome() - >>> out.value - [1] - >>> out.error - [ValueError('err')] - - """ - oks, errs = partition(list(self)) - try: - from .outcome import Outcome # noqa: PLC0415 - except ImportError: - raise ImportError( - "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) - - -class SafeStreamAsync[T, E](AsyncIterable["Result[T, E]"]): - """Async version of SafeStream for fallible asynchronous generators.""" - - def __init__(self, gen: AsyncIterator[Result[T, E]]) -> None: - """Initialize a SafeStreamAsync with a Result-yielding async iterator.""" - self._gen = gen - self._consumed = False - - 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 - - async def to_result(self) -> Result[list[T], E]: - """Transpose the async stream into a single Result (All-or-Nothing). - - Returns: - The combined Result of all yielded items. - - """ - items = [res async for res in self] - return combine(items) - - async def to_outcome(self) -> Outcome[list[T], list[E]]: - """Transpose the async stream into a fault-tolerant Outcome (Partial Success). - - Returns: - A master Outcome containing all success items and all errors. - - """ - items = [res async for res in self] - oks, errs = partition(items) - try: - from .outcome import Outcome # noqa: PLC0415 - except ImportError: - raise ImportError( - "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) - - -def _get_retry_delay( - current_delay: float, - *, - jitter: bool | float, -) -> float: - """Internal helper to calculate next retry delay.""" - if current_delay <= 0: - return 0 - sleep_time = current_delay - if jitter: - # If jitter is True, default to 0.1, else use the provided float value - jitter_val = jitter if isinstance(jitter, float) else 0.1 - sleep_time += random.uniform(0, jitter_val) - return sleep_time - - -@overload -def retry_result[T, E, **P]( - 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]]]: ... - - -@overload -def retry_result[T, E, **P]( - 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]]]: ... - - -@overload -def retry_result[T, E_exc: Exception, **P]( - 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]]]: ... - - -@overload -def retry_result[T, **P]( - 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]]]: ... - - -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]]: - # If catch is provided, we wrap the function in @catch first - wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) - - -@overload -def retry_result_async[T, E, **P]( - 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]]]]: ... - - -@overload -def retry_result_async[T, E, **P]( - 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]]]]: ... - - -@overload -def retry_result_async[T, E_exc: Exception, **P]( - 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]]]]: ... - - -@overload -def retry_result_async[T, **P]( - 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]]]]: ... - - -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 - wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) diff --git a/src/result/result.py b/src/result/result.py index a925bf0..ccf22c5 100644 --- a/src/result/result.py +++ b/src/result/result.py @@ -41,10 +41,15 @@ # pyright: reportPrivateUsage=false # pyright: reportOverlappingOverload=false # ruff: noqa: SLF001 +# mypy: disable-error-code="no-any-return" from __future__ import annotations +import asyncio import inspect +import random +import sys +import time from collections.abc import ( AsyncGenerator, Awaitable, @@ -687,7 +692,7 @@ def product[U, E2](self, other: Result[U, E2]) -> Result[tuple[T_co, U], E_co | """ if isinstance(other, Err): - return cast("Any", other) # type: ignore[no-any-return] # ty:ignore[unused-type-ignore-comment, unused-ignore-comment] + return cast("Any", other) return Ok[tuple[T_co, U]]((self._value, other._value)) @@ -1174,7 +1179,7 @@ def product[U, E2](self, _other: Result[U, E2]) -> Result[tuple[Any, U], E_co | Self unchanged. """ - return cast("Any", self) # type: ignore[no-any-return] # ty:ignore[unused-type-ignore-comment, unused-ignore-comment] + return cast("Any", self) OkErr: Final = (Ok, Err) @@ -2320,6 +2325,495 @@ def to_outcome[U](self, default: U) -> Outcome[U, E_co]: return Outcome(default, self._owner._error) +def _raise_assertion_error(message: str) -> Any: + """Internal helper to raise AssertionError and hide this frame from traceback.""" + __tracebackhide__ = True + try: + raise AssertionError(message) # noqa: TRY301 + except AssertionError as e: + tb = e.__traceback__ + raise e.with_traceback(tb.tb_next if tb else None) from None + + +class AssertOk: + """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. + + """ + + 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 + + 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 + + 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) + + def _trace_callback(self, frame: Any, event: str, _arg: Any) -> Any: + """Trace function that scans locals for Err variants after each line.""" + __tracebackhide__ = True + # Prevent re-entrancy issues if scanning logic itself triggers tracer + if self._is_scanning: + return self._trace_callback + + if event == "line": + self._is_scanning = True + try: + # Scan current local variables + for name, value in frame.f_locals.items(): + # Only check variables that were added during this block + if name not in self._initial_locals: + match value: + case Err(error_val): # pyright: ignore[reportUnknownVariableType] + # Trigger the error + error_str = cast("Any", error_val) + _raise_assertion_error(f"{self.message}: {error_str}") + case _: + pass + finally: + self._is_scanning = False + return self._trace_callback + + 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 + + +@overload +def assert_ok[T, E](result_or_message: Result[T, E]) -> T: ... + + +@overload +def assert_ok(result_or_message: str = "Result was Err") -> AssertOk: ... + + +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_decorator = catch + + +def _get_retry_delay( + current_delay: float, + *, + jitter: bool | float, +) -> float: + """Internal helper to calculate next retry delay.""" + if current_delay <= 0: + return 0 + sleep_time = current_delay + if jitter: + # If jitter is True, default to 0.1, else use the provided float value + jitter_val = jitter if isinstance(jitter, float) else 0.1 + sleep_time += random.uniform(0, jitter_val) + return sleep_time + + +@overload +def retry_result[T, E, **P]( + 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]]]: ... + + +@overload +def retry_result[T, E, **P]( + 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]]]: ... + + +@overload +def retry_result[T, E_exc: Exception, **P]( + 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]]]: ... + + +@overload +def retry_result[T, **P]( + 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]]]: ... + + +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]]: + # If catch is provided, we wrap the function in @catch first + wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) + + +@overload +def retry_result_async[T, E, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, E, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, E_exc: Exception, **P]( + 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]]]]: ... + + +@overload +def retry_result_async[T, **P]( + 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]]]]: ... + + +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 + wrapped_f = _catch_decorator(cast("Any", catch))(f) if 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) + + # --- Internal Teaching Helper --- diff --git a/tests/result/test_catch_adapters.py b/tests/result/test_catch_adapters.py index e0a7e14..2189606 100644 --- a/tests/result/test_catch_adapters.py +++ b/tests/result/test_catch_adapters.py @@ -82,8 +82,8 @@ def test_catch_instance_sync() -> None: assert safe.sync_ok(42) == Ok(42) res: Any = safe.sync_fail() - assert is_err(res) # ty: ignore[invalid-argument-type] - err: Any = res.err() # ty: ignore[unresolved-attribute] + assert is_err(res) # ty:ignore[invalid-argument-type] + err: Any = res.err() # ty:ignore[unresolved-attribute] assert isinstance(err, ValueError) assert str(err) == "sync fail" @@ -98,8 +98,8 @@ async def test_catch_instance_async() -> None: assert await safe.async_ok(100) == Ok(100) res: Any = await safe.async_fail() - assert is_err(res) # ty: ignore[invalid-argument-type] - err: Any = res.err() # ty: ignore[unresolved-attribute] + assert is_err(res) # ty:ignore[invalid-argument-type] + err: Any = res.err() # ty:ignore[unresolved-attribute] assert isinstance(err, ValueError) assert str(err) == "async fail" From 7354f71f052eb8398f91e4a95b06390f0fbb7fdc Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 13:03:50 +0800 Subject: [PATCH 7/8] Remove obsolete 'Future' documentation page The 'Future' documentation page has been removed as it is no longer relevant. The mkdocs.yml configuration has been updated to point to the 'adapters' page instead. --- docs/api/future.md | 5 ----- mkdocs.yml | 2 +- 2 files changed, 1 insertion(+), 6 deletions(-) delete mode 100644 docs/api/future.md diff --git a/docs/api/future.md b/docs/api/future.md deleted file mode 100644 index 9d37e88..0000000 --- a/docs/api/future.md +++ /dev/null @@ -1,5 +0,0 @@ -# Future - -Experimental and Alpha features for Result Pattern. - -::: result.future diff --git a/mkdocs.yml b/mkdocs.yml index 4f7ea2e..ac09e4b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -78,4 +78,4 @@ nav: - Result: api/result.md - Outcome: api/outcome.md - Combinators: api/combinators.md - - Future: api/future.md + - Future: api/adapters.md From fbeaebc0e498b855fe10664608a4308995707312 Mon Sep 17 00:00:00 2001 From: ZuidVolt Date: Thu, 12 Mar 2026 13:04:31 +0800 Subject: [PATCH 8/8] Rename 'Future' to 'Adapters' in docs nav --- mkdocs.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mkdocs.yml b/mkdocs.yml index ac09e4b..c2dea5a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -78,4 +78,4 @@ nav: - Result: api/result.md - Outcome: api/outcome.md - Combinators: api/combinators.md - - Future: api/adapters.md + - Adapters: api/adapters.md