diff --git a/README.md b/README.md index cf864ef..d6ff834 100644 --- a/README.md +++ b/README.md @@ -249,3 +249,83 @@ and returns internally stored `Maybe` instance. Also in some cases one might need point-free versions of `Maybe` interface methods, so one can access them via `maybe` module. For `FutureMaybe` point-free functions one can use `future_maybe` module. + +### `Result` + +Exception handling is one of the most important tasks in development. We often face +cases when invocation of some logic can lead to Exception raise. In python default +handling mechanism is `try` - `except` - `finally` block, which is actually just another +for of `if` statement. + +Worst thing about this approach is that commonly in python the only way to know that +function can raise an exception is documentation (which is not always written or written +good). There is no explicit mechanism to tell LSP / type checker / linter, that this +specific function needs exception handling. + +`Result` monad provides another approach, which is common for Rust and Go developers - +let's return exceptions instead of raising them. Thus we can explicitly tell that soma +action can fail and requires edge-case handling. + +Like `Maybe`, `Result` is just a protocol and has 2 implementations: + +- `Ok` - container indicating that calculation succeeded +- `Err` - container indicating that calculation failed + +Simplest case of using `Result` is division: + +```python +from wird import Result, Ok, Err + +def try_div(a: int, b: int) -> Result[float, ZeroDivisionError]: + if b == 0: + return Err(ZeroDivisionError()) + + return Ok(a / b) +``` + +There we explicitly tell that division operation can lead to failure and even pinpoint +specific type of error. + +`Result` provides the following interface: + +- `Result.unwrap` - extract internally stored value of `Ok` or raise `ErrUnwrapError` +- `Result.unwrap_or` - extract internally stored value of `Ok` or return other +- `Result.unwrap_or_else` - extract internally stored value of `Ok` or return closure + result +- `Result.unwrap_or_else_async` - same as `Result.unwrap_or_else`, but for async + closures +- `Result.unwrap_err` - same as `Result.unwrap`, but for `Err` +- `Result.unwrap_err_or` - same as `Result.unwrap_err_or`, but for `Err` +- `Result.unwrap_err_or_else` - same as `Result.unwrap_err_or_else`, but for `Err` +- `Result.unwrap_err_or_else_async` - same as `Result.unwrap_err_or_else_async`, but for + `Err` +- `Result.map` - binding method for `Ok` +- `Result.map_async` - same as `Result.map`, but for async functions +- `Result.inspect` - binding side-effect method for `Ok` +- `Result.inspect_async`- same as `Result.inspect_async`, but for async functions +- `Result.map_err` - same as `Result.map`, but for `Err` +- `Result.map_err_async` - same as `Result.map_async`, but for `Err` +- `Result.inspect_err` - same as `Result.inspect`, but for `Err` +- `Result.inspect_err_async` - same as `Result.inspect_async`, but for `Err` +- `Result.and_` - logical AND, replaces current `Result` with passed on `Ok` +- `Result.and_then` - same as `Result.map`, but for functions returning `Result` +- `Result.and_then_async` - same as `Result.and_then`, but for async functions +- `Result.or_` - logical OR, replaces current `Result` with passed on `Err` +- `Result.or_else` - same as `Result.map_err`, but for functions returning `Result` +- `Result.or_else_async` - same as `Result.or_else`, but for async functions +- `Result.is_ok` - `True` on `Ok` +- `Result.is_ok_and` - `True` on `Ok` and predicate `True` +- `Result.is_ok_and_async` - same as `Result.is_ok_and`, but for async predicate +- `Result.is_ok_or` - `True` on `Ok` or `Err` predicate `True` +- `Result.is_ok_or_async` - same as `Result.is_ok_or`, but for async predicate +- `Result.is_err` - `True` on `Err` +- `Result.is_err_and` - `True` on `Err` and predicate `True` +- `Result.is_err_and_async` - same as `Result.is_err_and`, but for async predicate +- `Result.is_err_or` - `True` on `Err` or `Ok` predicate `True +- `Result.is_err_or_async` - same as `result.is_err_or`, but for async predicate + +In the same manner as with `Maybe` we wird provides: + +- `FutureResult` as seamless adapter for `Future[Result]` +- point-free `Result` API in `wird.result` module +- point-free `FutureResult` API in `wird.future_result` module diff --git a/pyproject.toml b/pyproject.toml index bed7006..f7cbad3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,7 +36,13 @@ asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "module" [tool.coverage.run] -omit = ["tests/*", "wird/maybe.py", "wird/future_maybe.py"] +omit = [ + "tests/*", + "wird/maybe.py", + "wird/future_maybe.py", + "wird/result.py", + "wird/future_result.py", +] branch = true [tool.coverage.report] diff --git a/tests/test_maybe.py b/tests/test_maybe.py index e48038d..b6f24fe 100644 --- a/tests/test_maybe.py +++ b/tests/test_maybe.py @@ -2,8 +2,7 @@ import pytest -from wird import Empty, Future, Maybe, Some -from wird._maybe import EmptyUnwrapError, FutureMaybe +from wird import Empty, EmptyUnwrapError, Future, FutureMaybe, Maybe, Some async def test_some_unwraps() -> None: diff --git a/tests/test_result.py b/tests/test_result.py new file mode 100644 index 0000000..12335f8 --- /dev/null +++ b/tests/test_result.py @@ -0,0 +1,331 @@ +import operator + +import pytest + +from wird import Err, ErrUnwrapError, Future, FutureResult, Ok, OkUnwrapError, Result + + +async def test_ok_unwraps() -> None: + r = Ok(1).of_err_type(str) + + assert r.unwrap() == 1 + assert r.unwrap_or(2) == 1 + assert r.unwrap_or_else(lambda: 2) == 1 + assert await r.unwrap_or_else_async(lambda: Future.from_(2)) == 1 + + with pytest.raises(OkUnwrapError, match="expected Err, got Ok"): + r.unwrap_err() + + with pytest.raises(OkUnwrapError, match="on ok"): + r.unwrap_err(on_ok="on ok") + + assert r.unwrap_err_or("err") == "err" + assert r.unwrap_err_or_else(lambda: "err") == "err" + assert await r.unwrap_err_or_else_async(lambda: Future.from_("err")) == "err" + + assert await FutureResult.from_(r).unwrap() == 1 + assert await FutureResult.from_(r).unwrap_or(2) == 1 + assert await FutureResult.from_(r).unwrap_or_else(lambda: 2) == 1 + assert ( + await FutureResult.from_(r).unwrap_or_else_async(lambda: Future.from_(2)) == 1 + ) + + with pytest.raises(OkUnwrapError, match="expected Err, got Ok"): + await FutureResult.from_(r).unwrap_err() + + with pytest.raises(OkUnwrapError, match="on ok"): + await FutureResult.from_(r).unwrap_err(on_ok="on ok") + + assert await FutureResult.from_(r).unwrap_err_or("err") == "err" + assert await FutureResult.from_(r).unwrap_err_or_else(lambda: "err") == "err" + assert ( + await FutureResult.from_(r).unwrap_err_or_else_async( + lambda: Future.from_("err") + ) + == "err" + ) + + +async def test_err_unwraps() -> None: + r = Err("err").of_ok_type(int) + + with pytest.raises(ErrUnwrapError, match="expected Ok, got Err"): + r.unwrap() + + with pytest.raises(ErrUnwrapError, match="on err"): + r.unwrap(on_err="on err") + + assert r.unwrap_or(2) == 2 + assert r.unwrap_or_else(lambda: 2) == 2 + assert await r.unwrap_or_else_async(lambda: Future.from_(2)) == 2 + + assert r.unwrap_err() == "err" + assert r.unwrap_err_or("error") == "err" + assert r.unwrap_err_or_else(lambda: "error") == "err" + assert await r.unwrap_err_or_else_async(lambda: Future.from_("error")) == "err" + + with pytest.raises(ErrUnwrapError, match="expected Ok, got Err"): + await FutureResult.from_(r).unwrap() + + with pytest.raises(ErrUnwrapError, match="on err"): + await FutureResult.from_(r).unwrap(on_err="on err") + + assert await FutureResult.from_(r).unwrap_or(2) == 2 + assert await FutureResult.from_(r).unwrap_or_else(lambda: 2) == 2 + assert ( + await FutureResult.from_(r).unwrap_or_else_async(lambda: Future.from_(2)) == 2 + ) + + assert await FutureResult.from_(r).unwrap_err() == "err" + assert await FutureResult.from_(r).unwrap_err_or("error") == "err" + assert await FutureResult.from_(r).unwrap_err_or_else(lambda: "error") == "err" + assert ( + await FutureResult.from_(r).unwrap_err_or_else_async( + lambda: Future.from_("error") + ) + == "err" + ) + + +@pytest.mark.parametrize( + ("result", "target"), + [ + (Ok(1), Ok(5)), + (Err(""), Err("")), + ], +) +async def test_result_map( + result: Result[int, str], + target: Result[int, str], +) -> None: + assert ( + await result.map(operator.add, 1) + .map_async(lambda x: Future.from_(x + 1)) + .map(operator.add, 1) + .map_async(lambda x: Future.from_(x + 1)) + == target + ) + + +@pytest.mark.parametrize( + ("result", "target"), + [ + (Ok(""), Ok("")), + (Err(1), Err(5)), + ], +) +async def test_result_map_err( + result: Result[str, int], + target: Result[str, int], +) -> None: + assert ( + await result.map_err(operator.add, 1) + .map_err_async(lambda x: Future.from_(x + 1)) + .map_err(operator.add, 1) + .map_err_async(lambda x: Future.from_(x + 1)) + == target + ) + + +@pytest.mark.parametrize( + ("result", "target"), + [ + (Ok({}), Ok({"a": 1, "b": 2, "c": 3, "d": 4})), + (Err(""), Err("")), + ], +) +async def test_result_inspect( + result: Result[dict, str], + target: Result[dict, str], +) -> None: + async def set_item(data: dict, key: str, value: int) -> None: + data[key] = value + + assert ( + await result.inspect(operator.setitem, "a", 1) + .inspect_async(set_item, "b", 2) + .inspect(operator.setitem, "c", 3) + .inspect_async(set_item, "d", 4) + ) + + +@pytest.mark.parametrize( + ("result", "target"), + [ + (Err({}), Err({"a": 1, "b": 2, "c": 3, "d": 4})), + (Ok(""), Ok("")), + ], +) +async def test_result_inspect_err( + result: Result[str, dict], + target: Result[str, dict], +) -> None: + async def set_item(data: dict, key: str, value: int) -> None: + data[key] = value + + assert ( + await result.inspect_err(operator.setitem, "a", 1) + .inspect_err_async(set_item, "b", 2) + .inspect_err(operator.setitem, "c", 3) + .inspect_err_async(set_item, "d", 4) + ) == target + + +async def test_ok_predicates() -> None: + r = Ok(1).of_err_type(str) + + assert r.is_ok() + assert r.is_ok_and(lambda x: x % 2 == 1) + assert r.is_ok_or(lambda x: len(x) > 10) + assert await r.is_ok_and_async(lambda x: Future.from_(x % 2 == 1)) + assert await r.is_ok_or_async(lambda x: Future.from_(len(x) > 10)) + + assert not r.is_err() + assert not r.is_err_and(lambda x: len(x) > 10) + assert r.is_err_or(lambda x: x % 2 == 1) + assert not await r.is_err_and_async(lambda x: Future.from_(len(x) > 10)) + assert await r.is_err_or_async(lambda x: Future.from_(x % 2 == 1)) + + assert await FutureResult.from_(r).is_ok() + assert await FutureResult.from_(r).is_ok_and(lambda x: x % 2 == 1) + assert await FutureResult.from_(r).is_ok_or(lambda x: len(x) > 10) + assert await FutureResult.from_(r).is_ok_and_async( + lambda x: Future.from_(x % 2 == 1) + ) + assert await FutureResult.from_(r).is_ok_or_async( + lambda x: Future.from_(len(x) > 10) + ) + + assert not await FutureResult.from_(r).is_err() + assert not await FutureResult.from_(r).is_err_and(lambda x: len(x) > 10) + assert await FutureResult.from_(r).is_err_or(lambda x: x % 2 == 1) + assert not await FutureResult.from_(r).is_err_and_async( + lambda x: Future.from_(len(x) > 10) + ) + assert await FutureResult.from_(r).is_err_or_async( + lambda x: Future.from_(x % 2 == 1) + ) + + +async def test_err_predicates() -> None: + r = Err(1).of_ok_type(str) + + assert not r.is_ok() + assert not r.is_ok_and(lambda x: len(x) > 10) + assert r.is_ok_or(lambda x: x % 2 == 1) + assert not await r.is_ok_and_async(lambda x: Future.from_(len(x) > 10)) + assert await r.is_ok_or_async(lambda x: Future.from_(x % 2 == 1)) + + assert r.is_err() + assert r.is_err_and(lambda x: x % 2 == 1) + assert r.is_err_or(lambda x: len(x) > 10) + assert await r.is_err_and_async(lambda x: Future.from_(x % 2 == 1)) + assert await r.is_err_or_async(lambda x: Future.from_(len(x) > 10)) + + assert not await FutureResult.from_(r).is_ok() + assert not await FutureResult.from_(r).is_ok_and(lambda x: len(x) > 10) + assert await FutureResult.from_(r).is_ok_or(lambda x: x % 2 == 1) + assert not await FutureResult.from_(r).is_ok_and_async( + lambda x: Future.from_(len(x) > 10) + ) + assert await FutureResult.from_(r).is_ok_or_async( + lambda x: Future.from_(x % 2 == 1) + ) + + assert await FutureResult.from_(r).is_err() + assert await FutureResult.from_(r).is_err_and(lambda x: x % 2 == 1) + assert await FutureResult.from_(r).is_err_or(lambda x: len(x) > 10) + assert await FutureResult.from_(r).is_err_and_async( + lambda x: Future.from_(x % 2 == 1) + ) + assert await FutureResult.from_(r).is_err_or_async( + lambda x: Future.from_(len(x) > 10) + ) + + +async def test_ok_and() -> None: + r = Ok(1).of_err_type(str) + + assert r.and_(Ok(2)) == Ok(2) + assert r.and_(Err("")) == Err("") + assert r.and_then(lambda _: Ok(2)) == Ok(2) + assert r.and_then(lambda _: Err("")) == Err("") + assert await r.and_then_async(lambda _: Future.from_(Ok(2))) == Ok(2) + assert await r.and_then_async(lambda _: Future.from_(Err(""))) == Err("") + + assert await FutureResult.from_(r).and_(Ok(2)) == Ok(2) + assert await FutureResult.from_(r).and_(Err("")) == Err("") + assert await FutureResult.from_(r).and_then(lambda _: Ok(2)) == Ok(2) + assert await FutureResult.from_(r).and_then(lambda _: Err("")) == Err("") + assert await FutureResult.from_(r).and_then_async( + lambda _: Future.from_(Ok(2)) + ) == Ok(2) + assert await FutureResult.from_(r).and_then_async( + lambda _: Future.from_(Err("")) + ) == Err("") + + +async def test_ok_or() -> None: + r = Ok(1).of_err_type(str) + + assert r.or_(Ok(2)) == Ok(1) + assert r.or_(Err("")) == Ok(1) + assert r.or_else(lambda _: Ok(2)) == Ok(1) + assert r.or_else(lambda _: Err("")) == Ok(1) + assert await r.or_else_async(lambda _: Future.from_(Ok(2))) == Ok(1) + assert await r.or_else_async(lambda _: Future.from_(Err(""))) == Ok(1) + + assert await FutureResult.from_(r).or_(Ok(2)) == Ok(1) + assert await FutureResult.from_(r).or_(Err("")) == Ok(1) + assert await FutureResult.from_(r).or_else(lambda _: Ok(2)) == Ok(1) + assert await FutureResult.from_(r).or_else(lambda _: Err("")) == Ok(1) + assert await FutureResult.from_(r).or_else_async( + lambda _: Future.from_(Ok(2)) + ) == Ok(1) + assert await FutureResult.from_(r).or_else_async( + lambda _: Future.from_(Err("")) + ) == Ok(1) + + +async def test_err_and() -> None: + r = Err(1).of_ok_type(str) + + assert r.and_(Ok("ok")) == Err(1) + assert r.and_(Err(2)) == Err(1) + assert r.and_then(lambda _: Ok("ok")) == Err(1) + assert r.and_then(lambda _: Err(2)) == Err(1) + assert await r.and_then_async(lambda _: Future.from_(Ok("ok"))) == Err(1) + assert await r.and_then_async(lambda _: Future.from_(Err(2))) == Err(1) + + assert await FutureResult.from_(r).and_(Ok("ok")) == Err(1) + assert await FutureResult.from_(r).and_(Err(2)) == Err(1) + assert await FutureResult.from_(r).and_then(lambda _: Ok("ok")) == Err(1) + assert await FutureResult.from_(r).and_then(lambda _: Err(2)) == Err(1) + assert await FutureResult.from_(r).and_then_async( + lambda _: Future.from_(Ok("ok")) + ) == Err(1) + assert await FutureResult.from_(r).and_then_async( + lambda _: Future.from_(Err(2)) + ) == Err(1) + + +async def test_err_or() -> None: + r = Err(1).of_ok_type(str) + + assert r.or_(Ok("ok")) == Ok("ok") + assert r.or_(Err(2)) == Err(2) + assert r.or_else(lambda _: Ok("ok")) == Ok("ok") + assert r.or_else(lambda _: Err(2)) == Err(2) + assert await r.or_else_async(lambda _: Future.from_(Ok("ok"))) == Ok("ok") + assert await r.or_else_async(lambda _: Future.from_(Err(2))) == Err(2) + + assert await FutureResult.from_(r).or_(Ok("ok")) == Ok("ok") + assert await FutureResult.from_(r).or_(Err(2)) == Err(2) + assert await FutureResult.from_(r).or_else(lambda _: Ok("ok")) == Ok("ok") + assert await FutureResult.from_(r).or_else(lambda _: Err(2)) == Err(2) + assert await FutureResult.from_(r).or_else_async( + lambda _: Future.from_(Ok("ok")) + ) == Ok("ok") + assert await FutureResult.from_(r).or_else_async( + lambda _: Future.from_(Err(2)) + ) == Err(2) diff --git a/wird/__init__.py b/wird/__init__.py index 5b3fd88..63a85d9 100644 --- a/wird/__init__.py +++ b/wird/__init__.py @@ -1,16 +1,26 @@ from . import maybe, future_maybe # noqa +from . import result, future_result from ._future import Future from ._maybe import Empty, EmptyUnwrapError, FutureMaybe, Maybe, Some from ._value import Value +from ._result import Result, Err, Ok, FutureResult, OkUnwrapError, ErrUnwrapError __all__ = ( "Empty", "EmptyUnwrapError", + "Err", + "ErrUnwrapError", "Future", "FutureMaybe", + "FutureResult", "Maybe", + "Ok", + "OkUnwrapError", + "Result", "Some", "Value", "future_maybe", + "future_result", "maybe", + "result", ) diff --git a/wird/_maybe.py b/wird/_maybe.py index 48339ef..9a68f23 100644 --- a/wird/_maybe.py +++ b/wird/_maybe.py @@ -6,6 +6,7 @@ Awaitable, Callable, Concatenate, + Generator, NoReturn, Protocol, Type, @@ -653,7 +654,7 @@ def __init__(self, internal: Awaitable[Maybe[T]]) -> None: def from_[V](value: Maybe[V]) -> FutureMaybe[V]: return FutureMaybe(f.Future.from_(value)) - def __await__(self) -> f.Generator[Any, Any, Maybe[T]]: + def __await__(self) -> Generator[Any, Any, Maybe[T]]: return self.internal.__await__() @overload diff --git a/wird/_result.py b/wird/_result.py new file mode 100644 index 0000000..f202eda --- /dev/null +++ b/wird/_result.py @@ -0,0 +1,1347 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import ( + Any, + Awaitable, + Callable, + Concatenate, + Generator, + Protocol, + Type, + overload, +) + +from . import _future as f, result + +__all__ = ( + "Err", + "ErrUnwrapError", + "FutureResult", + "Ok", + "OkUnwrapError", + "Result", +) + + +class ErrUnwrapError(ValueError): ... + + +class OkUnwrapError(ValueError): ... + + +class Result[T, E](Protocol): # pragma: no cover + @overload + def unwrap[R]( + self, + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", + ) -> R: + """Returns the contained Ok value casted to passed type. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + @overload + def unwrap(self, *, on_err: str = "expected Ok, got Err") -> T: + """Returns the contained Ok value. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + """ + ... + + def unwrap_or(self, /, other: T) -> T: + """Returns the contained Ok value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_or_else, which is + lazily evaluated. + """ + ... + + def unwrap_or_else[**P]( + self, + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, + ) -> T: + """Returns the contained Ok value or computes it from a closure.""" + ... + + def unwrap_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[T]: + """Returns the contained Ok value or computes it from an async closure.""" + ... + + @overload + def unwrap_err[R]( + self, + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", + ) -> R: + """Returns the contained Err value casted to passed type. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + @overload + def unwrap_err(self, *, on_ok: str = "expected Err, got Ok") -> E: + """Returns the contained Ok value. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + """ + ... + + def unwrap_err_or(self, /, other: E) -> E: + """Returns the contained Err value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_err_or_else, which is + lazily evaluated. + """ + ... + + def unwrap_err_or_else[**P]( + self, + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, + ) -> E: + """Returns the contained Err value or computes it from a closure.""" + ... + + def unwrap_err_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[E]: + """Returns the contained Err value or computes it from an async closure.""" + ... + + def map[**P, R]( + self, + fn: Callable[Concatenate[T, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, E]: + """Maps a Result[T, E] to Result[R, E] by applying a function to a contained Ok + value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + ... + + def map_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Maps a Result[T, E] to FutureResult[R, E] by applying an async function to a + contained Ok value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + ... + + def inspect[**P]( + self, + fn: Callable[Concatenate[T, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, E]: + """Calls a function with a reference to the contained value if Ok. + + Returns the original result. + """ + ... + + def inspect_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + ... + + def map_err[**P, R]( + self, + fn: Callable[Concatenate[E, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, R]: + """Maps a Result[T, E] to Result[T, R] by applying a function to a contained Err + value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + ... + + def map_err_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Maps a Result[T, E] to FutureResult[T, R] by applying am async function to a + contained Err value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + ... + + def inspect_err[**P]( + self, + fn: Callable[Concatenate[E, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, E]: + """Calls a function with a reference to the contained value if Err. + + Returns the original result. + """ + ... + + def inspect_err_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + ... + + def and_[R](self, other: Result[R, E]) -> Result[R, E]: + """Returns other if the result is Ok, otherwise returns the Err value of self. + + Arguments passed to and are eagerly evaluated; if you are passing the result of + a function call, it is recommended to use and_then, which is lazily evaluated. + """ + ... + + def and_then[**P, R]( + self, + fn: Callable[Concatenate[T, P], Result[R, E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, E]: + """Calls a function if the result is Ok, otherwise returns the Err value of + self. + + This function can be used for control flow based on Result values. + """ + ... + + def and_then_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Result[R, E]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Calls an async function if the result is Ok, otherwise returns the Err value + of self. + + This function can be used for control flow based on Result values. + """ + ... + + def or_[R](self, other: Result[T, R]) -> Result[T, R]: + """Returns other if the result is Err, otherwise returns the Ok value of self. + + Arguments passed to or are eagerly evaluated; if you are passing the result of a + function call, it is recommended to use or_else, which is lazily evaluated. + """ + ... + + def or_else[**P, R]( + self, + fn: Callable[Concatenate[E, P], Result[T, R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, R]: + """Calls a function if the result is Err, otherwise returns the Ok value of + self. + + This function can be used for control flow based on result values. + """ + ... + + def or_else_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Result[T, R]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Calls an async function if the result is Err, otherwise returns the Ok value + of self. + + This function can be used for control flow based on result values. + """ + ... + + def is_ok(self) -> bool: + """Returns True if the result is Ok.""" + ... + + def is_ok_and[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + """Returns True if the result is Ok and the value inside of it matches a + predicate.""" + ... + + def is_ok_and_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches an async + predicate.""" + ... + + def is_ok_or[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + """Returns True if the result is Ok or the value inside of Err matches a + predicate.""" + ... + + def is_ok_or_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches an async + predicate.""" + ... + + def is_err(self) -> bool: + """Returns True if the result is Err.""" + ... + + def is_err_and[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + """Returns True if the result is Err and the value inside of it matches a + predicate.""" + ... + + def is_err_and_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches an async + predicate.""" + ... + + def is_err_or[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + """Returns True if the result is Err or the value inside of Ok matches a + predicate.""" + ... + + def is_err_or_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches an async + predicate.""" + ... + + +@dataclass(slots=True, frozen=True) +class Ok[T](Result[T, Any]): + internal: T + + def of_err_type[E](self, _: Type[E]) -> Result[T, E]: + return self + + @overload + def unwrap[R]( + self, + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", + ) -> R: ... + + @overload + def unwrap(self, *, on_err: str = "expected Ok, got Err") -> T: ... + + def unwrap(self, **kwargs) -> Any: + return self.internal + + def unwrap_or(self, /, other: T) -> T: + return self.internal + + def unwrap_or_else[**P]( + self, + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, + ) -> T: + return self.internal + + def unwrap_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[T]: + return f.Future.from_(self.internal) + + @overload + def unwrap_err[R]( + self, + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", + ) -> R: ... + + @overload + def unwrap_err(self, *, on_ok: str = "expected Err, got Ok") -> T: ... + + def unwrap_err(self, **kwargs) -> Any: + raise OkUnwrapError(kwargs.get("on_ok", "expected Err, got Ok")) + + def unwrap_err_or[E](self, /, other: E) -> E: + return other + + def unwrap_err_or_else[E, **P]( + self, + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, + ) -> E: + return fn(*args, **kwargs) + + def unwrap_err_or_else_async[E, **P]( + self, + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[E]: + return f.Future(fn(*args, **kwargs)) + + def map[**P, R]( + self, + fn: Callable[Concatenate[T, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, Any]: + return Ok(fn(self.internal, *args, **kwargs)) + + def map_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, Any]: + return FutureResult(_ok_map_async(self.internal, fn, *args, **kwargs)) + + def inspect[**P]( + self, + fn: Callable[Concatenate[T, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, Any]: + fn(self.internal, *args, **kwargs) + return self + + def inspect_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, Any]: + return FutureResult(_ok_inspect_async(self.internal, fn, *args, **kwargs)) + + def map_err[**P, R]( + self, + fn: Callable[Concatenate[Any, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, R]: + return self + + def map_err_async[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + return FutureResult.from_(self) + + def inspect_err[**P]( + self, + fn: Callable[Concatenate[Any, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, Any]: + return self + + def inspect_err_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, Any]: + return FutureResult.from_(self) + + def and_[R](self, other: Result[R, Any]) -> Result[R, Any]: + return other + + def and_then[**P, R]( + self, + fn: Callable[Concatenate[T, P], Result[R, Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, Any]: + return fn(self.internal, *args, **kwargs) + + def and_then_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Result[R, Any]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, Any]: + return FutureResult(fn(self.internal, *args, **kwargs)) + + def or_[R](self, other: Result[T, R]) -> Result[T, R]: + return self + + def or_else[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Result[T, R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[T, R]: + return self + + def or_else_async[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[Result[T, R]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + return FutureResult.from_(self) + + def is_ok(self) -> bool: + return True + + def is_ok_and[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return fn(self.internal, *args, **kwargs) + + def is_ok_and_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future(fn(self.internal, *args, **kwargs)) + + def is_ok_or[**P]( + self, + fn: Callable[Concatenate[Any, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return True + + def is_ok_or_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future.from_(True) + + def is_err(self) -> bool: + return False + + def is_err_and[**P]( + self, + fn: Callable[Concatenate[Any, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return False + + def is_err_and_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future.from_(False) + + def is_err_or[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return fn(self.internal, *args, **kwargs) + + def is_err_or_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future(fn(self.internal, *args, **kwargs)) + + +async def _ok_map_async[T, **P, R]( + value: T, + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> Result[R, Any]: + return Ok(await fn(value, *args, **kwargs)) + + +async def _ok_inspect_async[T, **P]( + value: T, + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> Result[T, Any]: + await fn(value, *args, **kwargs) + return Ok(value) + + +@dataclass(slots=True, frozen=True) +class Err[E](Result[Any, E]): + internal: E + + def of_ok_type[T](self, _: Type[T]) -> Result[T, E]: + return self + + @overload + def unwrap[R]( + self, + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", + ) -> R: ... + + @overload + def unwrap(self, *, on_err: str = "expected Ok, got Err") -> Any: ... + + def unwrap(self, **kwargs) -> Any: + raise ErrUnwrapError(kwargs.get("on_err", "expected Ok, got Err")) + + def unwrap_or[T](self, /, other: T) -> T: + return other + + def unwrap_or_else[T, **P]( + self, + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, + ) -> T: + return fn(*args, **kwargs) + + def unwrap_or_else_async[T, **P]( + self, + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[T]: + return f.Future(fn(*args, **kwargs)) + + @overload + def unwrap_err[R]( + self, + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", + ) -> R: ... + + @overload + def unwrap_err(self, *, on_ok: str = "expected Err, got Ok") -> E: ... + + def unwrap_err(self, **kwargs) -> Any: + return self.internal + + def unwrap_err_or(self, /, other: E) -> E: + return self.internal + + def unwrap_err_or_else[**P]( + self, + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, + ) -> E: + return self.internal + + def unwrap_err_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[E]: + return f.Future.from_(self.internal) + + def map[**P, R]( + self, + fn: Callable[Concatenate[Any, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, E]: + return self + + def map_async[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + return FutureResult.from_(self) + + def inspect[**P]( + self, + fn: Callable[Concatenate[Any, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[Any, E]: + return self + + def inspect_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[Any, E]: + return FutureResult.from_(self) + + def map_err[**P, R]( + self, + fn: Callable[Concatenate[E, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[Any, R]: + return Err(fn(self.internal, *args, **kwargs)) + + def map_err_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[Any, R]: + return FutureResult(_err_map_err_async(self.internal, fn, *args, **kwargs)) + + def inspect_err[**P]( + self, + fn: Callable[Concatenate[E, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[Any, E]: + fn(self.internal, *args, **kwargs) + return self + + def inspect_err_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[Any, E]: + return FutureResult(_err_inspect_err_async(self.internal, fn, *args, **kwargs)) + + def and_[R](self, other: Result[R, E]) -> Result[R, E]: + return self + + def and_then[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Result[R, E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[R, E]: + return self + + def and_then_async[**P, R]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[Result[R, E]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + return FutureResult.from_(self) + + def or_[R](self, other: Result[Any, R]) -> Result[Any, R]: + return other + + def or_else[**P, R]( + self, + fn: Callable[Concatenate[E, P], Result[Any, R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> Result[Any, R]: + return fn(self.internal, *args, **kwargs) + + def or_else_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Result[Any, R]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[Any, R]: + return FutureResult(fn(self.internal, *args, **kwargs)) + + def is_ok(self) -> bool: + return False + + def is_ok_and[**P]( + self, + fn: Callable[Concatenate[Any, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return False + + def is_ok_and_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future.from_(False) + + def is_ok_or[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return fn(self.internal, *args, **kwargs) + + def is_ok_or_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future(fn(self.internal, *args, **kwargs)) + + def is_err(self) -> bool: + return True + + def is_err_and[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return fn(self.internal, *args, **kwargs) + + def is_err_and_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future(fn(self.internal, *args, **kwargs)) + + def is_err_or[**P]( + self, + fn: Callable[Concatenate[Any, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> bool: + return True + + def is_err_or_async[**P]( + self, + fn: Callable[Concatenate[Any, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + return f.Future.from_(True) + + +async def _err_map_err_async[E, **P, R]( + value: E, + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> Err[R]: + return Err(await fn(value, *args, **kwargs)) + + +async def _err_inspect_err_async[E, **P]( + value: E, + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> Err[E]: + await fn(value, *args, **kwargs) + return Err(value) + + +@dataclass(slots=True, frozen=True) +class FutureResult[T, E]: + internal: f.Future[Result[T, E]] + + def __init__(self, internal: Awaitable[Result[T, E]]) -> None: + object.__setattr__( + self, + "internal", + f.Future(internal) if not isinstance(internal, f.Future) else internal, + ) + + @staticmethod + def from_[V, U](value: Result[V, U]) -> FutureResult[V, U]: + return FutureResult(f.Future.from_(value)) + + def __await__(self) -> Generator[Any, Any, Result[T, E]]: + return self.internal.__await__() + + @overload + def unwrap[R]( + self, + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", + ) -> f.Future[R]: + """Returns the contained Ok value casted to passed type. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + @overload + def unwrap(self, *, on_err: str = "expected Ok, got Err") -> f.Future[T]: + """Returns the contained Ok value. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + """ + ... + + def unwrap(self, **kwargs) -> f.Future[Any]: + return self.internal.map(result.unwrap, **kwargs) # type: ignore[arg-type] + + def unwrap_or(self, /, other: T) -> f.Future[T]: + """Returns the contained Ok value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_or_else, which is + lazily evaluated. + """ + return self.internal.map(result.unwrap_or, other) + + def unwrap_or_else[**P]( + self, + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[T]: + """Returns the contained Ok value or computes it from a closure.""" + return self.internal.map(result.unwrap_or_else, fn, *args, **kwargs) + + def unwrap_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[T]: + """Returns the contained Ok value or computes it from an async closure.""" + return self.internal.map_async(result.unwrap_or_else_async, fn, *args, **kwargs) + + @overload + def unwrap_err[R]( + self, + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", + ) -> f.Future[R]: + """Returns the contained Err value casted to passed type. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + @overload + def unwrap_err(self, *, on_ok: str = "expected Err, got Ok") -> f.Future[E]: + """Returns the contained Ok value. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + """ + ... + + def unwrap_err(self, **kwargs) -> f.Future[Any]: + return self.internal.map(result.unwrap_err, **kwargs) # type: ignore[arg-type] + + def unwrap_err_or(self, /, other: E) -> f.Future[E]: + """Returns the contained Err value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_err_or_else, which is + lazily evaluated. + """ + return self.internal.map(result.unwrap_err_or, other) + + def unwrap_err_or_else[**P]( + self, + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[E]: + """Returns the contained Err value or computes it from a closure.""" + return self.internal.map(result.unwrap_err_or_else, fn, *args, **kwargs) + + def unwrap_err_or_else_async[**P]( + self, + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[E]: + """Returns the contained Err value or computes it from an async closure.""" + return self.internal.map_async( + result.unwrap_err_or_else_async, fn, *args, **kwargs + ) + + def map[**P, R]( + self, + fn: Callable[Concatenate[T, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Maps a Result[T, E] to Result[R, E] by applying a function to a contained Ok + value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return FutureResult(self.internal.map(result.map, fn, *args, **kwargs)) + + def map_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Maps a Result[T, E] to FutureResult[R, E] by applying an async function to a + contained Ok value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return FutureResult( + self.internal.map_async(result.map_async, fn, *args, **kwargs) + ) + + def inspect[**P]( + self, + fn: Callable[Concatenate[T, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls a function with a reference to the contained value if Ok. + + Returns the original result. + """ + return FutureResult(self.internal.map(result.inspect, fn, *args, **kwargs)) + + def inspect_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + return FutureResult( + self.internal.map_async(result.inspect_async, fn, *args, **kwargs) + ) + + def map_err[**P, R]( + self, + fn: Callable[Concatenate[E, P], R], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Maps a Result[T, E] to Result[T, R] by applying a function to a contained Err + value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return FutureResult(self.internal.map(result.map_err, fn, *args, **kwargs)) + + def map_err_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Maps a Result[T, E] to FutureResult[T, R] by applying am async function to a + contained Err value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return FutureResult( + self.internal.map_async(result.map_err_async, fn, *args, **kwargs) + ) + + def inspect_err[**P]( + self, + fn: Callable[Concatenate[E, P], Any], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls a function with a reference to the contained value if Err. + + Returns the original result. + """ + return FutureResult(self.internal.map(result.inspect_err, fn, *args, **kwargs)) + + def inspect_err_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + return FutureResult( + self.internal.map_async(result.inspect_err_async, fn, *args, **kwargs) + ) + + def and_[R](self, other: Result[R, E]) -> FutureResult[R, E]: + """Returns other if the result is Ok, otherwise returns the Err value of self. + + Arguments passed to and are eagerly evaluated; if you are passing the result of + a function call, it is recommended to use and_then, which is lazily evaluated. + """ + return FutureResult(self.internal.map(result.and_, other)) + + def and_then[**P, R]( + self, + fn: Callable[Concatenate[T, P], Result[R, E]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Calls a function if the result is Ok, otherwise returns the Err value of + self. + + This function can be used for control flow based on Result values. + """ + return FutureResult(self.internal.map(result.and_then, fn, *args, **kwargs)) + + def and_then_async[**P, R]( + self, + fn: Callable[Concatenate[T, P], Awaitable[Result[R, E]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[R, E]: + """Calls an async function if the result is Ok, otherwise returns the Err value + of self. + + This function can be used for control flow based on Result values. + """ + return FutureResult( + self.internal.map_async(result.and_then_async, fn, *args, **kwargs) + ) + + def or_[R](self, other: Result[T, R]) -> FutureResult[T, R]: + """Returns other if the result is Err, otherwise returns the Ok value of self. + + Arguments passed to or are eagerly evaluated; if you are passing the result of a + function call, it is recommended to use or_else, which is lazily evaluated. + """ + return FutureResult(self.internal.map(result.or_, other)) + + def or_else[**P, R]( + self, + fn: Callable[Concatenate[E, P], Result[T, R]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Calls a function if the result is Err, otherwise returns the Ok value of + self. + + This function can be used for control flow based on result values. + """ + return FutureResult(self.internal.map(result.or_else, fn, *args, **kwargs)) + + def or_else_async[**P, R]( + self, + fn: Callable[Concatenate[E, P], Awaitable[Result[T, R]]], + *args: P.args, + **kwargs: P.kwargs, + ) -> FutureResult[T, R]: + """Calls an async function if the result is Err, otherwise returns the Ok value + of self. + + This function can be used for control flow based on result values. + """ + return FutureResult( + self.internal.map_async(result.or_else_async, fn, *args, **kwargs) + ) + + def is_ok(self) -> f.Future[bool]: + """Returns True if the result is Ok.""" + return self.internal.map(result.is_ok) + + def is_ok_and[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches a + predicate.""" + return self.internal.map(result.is_ok_and, fn, *args, **kwargs) + + def is_ok_and_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches an async + predicate.""" + return self.internal.map_async(result.is_ok_and_async, fn, *args, **kwargs) + + def is_ok_or[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches a + predicate.""" + return self.internal.map(result.is_ok_or, fn, *args, **kwargs) + + def is_ok_or_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches an async + predicate.""" + return self.internal.map_async(result.is_ok_or_async, fn, *args, **kwargs) + + def is_err(self) -> f.Future[bool]: + """Returns True if the result is Err.""" + return self.internal.map(result.is_err) + + def is_err_and[**P]( + self, + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches a + predicate.""" + return self.internal.map(result.is_err_and, fn, *args, **kwargs) + + def is_err_and_async[**P]( + self, + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches an async + predicate.""" + return self.internal.map_async(result.is_err_and_async, fn, *args, **kwargs) + + def is_err_or[**P]( + self, + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches a + predicate.""" + return self.internal.map(result.is_err_or, fn, *args, **kwargs) + + def is_err_or_async[**P]( + self, + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, + ) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches an async + predicate.""" + return self.internal.map_async(result.is_err_or_async, fn, *args, **kwargs) diff --git a/wird/future_result.py b/wird/future_result.py new file mode 100644 index 0000000..0fe690c --- /dev/null +++ b/wird/future_result.py @@ -0,0 +1,460 @@ +from typing import Any, Awaitable, Callable, Concatenate, Type, overload + +from . import _future as f, _result as r + +__all__ = ( + "and_", + "and_then", + "and_then_async", + "inspect", + "inspect_async", + "inspect_err", + "inspect_err_async", + "is_err", + "is_err_and", + "is_err_and_async", + "is_err_or", + "is_err_or_async", + "is_ok", + "is_ok_and", + "is_ok_and_async", + "is_ok_or", + "is_ok_or_async", + "map", + "map_async", + "map_err", + "map_err_async", + "or_", + "or_else", + "or_else_async", + "unwrap", + "unwrap_err", + "unwrap_err_or", + "unwrap_err_or_else", + "unwrap_err_or_else_async", + "unwrap_or", + "unwrap_or_else", + "unwrap_or_else_async", +) + + +@overload +def unwrap[T, E, R]( + result: r.FutureResult[T, E], + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", +) -> f.Future[R]: + """Returns the contained Ok value casted to passed type. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + +@overload +def unwrap[T, E]( + result: r.FutureResult[T, E], *, on_err: str = "expected Ok, got Err" +) -> f.Future[T]: + """Returns the contained Ok value. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + """ + ... + + +def unwrap[T, E](result: r.FutureResult[T, E], **kwargs) -> f.Future[Any]: + return result.unwrap(**kwargs) + + +def unwrap_or[T, E](result: r.FutureResult[T, E], /, other: T) -> f.Future[T]: + """Returns the contained Ok value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_or_else, which is + lazily evaluated. + """ + return result.unwrap_or(other) + + +def unwrap_or_else[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[T]: + """Returns the contained Ok value or computes it from a closure.""" + return result.unwrap_or_else(fn, *args, **kwargs) + + +def unwrap_or_else_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[T]: + """Returns the contained Ok value or computes it from an async closure.""" + return result.unwrap_or_else_async(fn, *args, **kwargs) + + +@overload +def unwrap_err[T, E, R]( + result: r.FutureResult[T, E], + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", +) -> f.Future[R]: + """Returns the contained Err value casted to passed type. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + +@overload +def unwrap_err[T, E]( + result: r.FutureResult[T, E], *, on_ok: str = "expected Err, got Ok" +) -> f.Future[E]: + """Returns the contained Ok value. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + """ + ... + + +def unwrap_err[T, E](result: r.FutureResult[T, E], **kwargs) -> f.Future[Any]: + return result.unwrap_err(**kwargs) + + +def unwrap_err_or[T, E](result: r.FutureResult[T, E], /, other: E) -> f.Future[E]: + """Returns the contained Err value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_err_or_else, which is + lazily evaluated. + """ + return result.unwrap_err_or(other) + + +def unwrap_err_or_else[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[E]: + """Returns the contained Err value or computes it from a closure.""" + return result.unwrap_err_or_else(fn, *args, **kwargs) + + +def unwrap_err_or_else_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[E]: + """Returns the contained Err value or computes it from an async closure.""" + return result.unwrap_err_or_else_async(fn, *args, **kwargs) + + +def map[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], R], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Maps a Result[T, E] to Result[R, E] by applying a function to a contained Ok + value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return result.map(fn, *args, **kwargs) + + +def map_async[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Maps a Result[T, E] to FutureResult[R, E] by applying an async function to a + contained Ok value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return result.map_async(fn, *args, **kwargs) + + +def inspect[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Any], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls a function with a reference to the contained value if Ok. + + Returns the original result. + """ + return result.inspect(fn, *args, **kwargs) + + +def inspect_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + return result.inspect_async(fn, *args, **kwargs) + + +def map_err[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], R], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Maps a Result[T, E] to Result[T, R] by applying a function to a contained Err + value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return result.map_err(fn, *args, **kwargs) + + +def map_err_async[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Maps a Result[T, E] to FutureResult[T, R] by applying am async function to a + contained Err value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return result.map_err_async(fn, *args, **kwargs) + + +def inspect_err[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Any], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls a function with a reference to the contained value if Err. + + Returns the original result. + """ + return result.inspect_err(fn, *args, **kwargs) + + +def inspect_err_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as FutureResult. + """ + return result.inspect_err_async(fn, *args, **kwargs) + + +def and_[T, E, R]( + result: r.FutureResult[T, E], other: r.Result[R, E] +) -> r.FutureResult[R, E]: + """Returns other if the result is Ok, otherwise returns the Err value of result + + Arguments passed to and are eagerly evaluated; if you are passing the result of + a function call, it is recommended to use and_then, which is lazily evaluated. + """ + return result.and_(other) + + +def and_then[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], r.Result[R, E]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Calls a function if the result is Ok, otherwise returns the Err value of + result + + This function can be used for control flow based on Result values. + """ + return result.and_then(fn, *args, **kwargs) + + +def and_then_async[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Awaitable[r.Result[R, E]]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Calls an async function if the result is Ok, otherwise returns the Err value + of result + + This function can be used for control flow based on Result values. + """ + return result.and_then_async(fn, *args, **kwargs) + + +def or_[T, E, R]( + result: r.FutureResult[T, E], other: r.Result[T, R] +) -> r.FutureResult[T, R]: + """Returns other if the result is Err, otherwise returns the Ok value of result + + Arguments passed to or are eagerly evaluated; if you are passing the result of a + function call, it is recommended to use or_else, which is lazily evaluated. + """ + return result.or_(other) + + +def or_else[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], r.Result[T, R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Calls a function if the result is Err, otherwise returns the Ok value of + result + + This function can be used for control flow based on result values. + """ + return result.or_else(fn, *args, **kwargs) + + +def or_else_async[T, E, **P, R]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Awaitable[r.Result[T, R]]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Calls an async function if the result is Err, otherwise returns the Ok value + of result + + This function can be used for control flow based on result values. + """ + return result.or_else_async(fn, *args, **kwargs) + + +def is_ok[T, E](result: r.FutureResult[T, E]) -> f.Future[bool]: + """Returns True if the result is Ok.""" + return result.is_ok() + + +def is_ok_and[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches a + predicate.""" + return result.is_ok_and(fn, *args, **kwargs) + + +def is_ok_and_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches an async + predicate.""" + return result.is_ok_and_async(fn, *args, **kwargs) + + +def is_ok_or[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches a + predicate.""" + return result.is_ok_or(fn, *args, **kwargs) + + +def is_ok_or_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches an async + predicate.""" + return result.is_ok_or_async(fn, *args, **kwargs) + + +def is_err[T, E](result: r.FutureResult[T, E]) -> f.Future[bool]: + """Returns True if the result is Err.""" + return result.is_err() + + +def is_err_and[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches a + predicate.""" + return result.is_err_and(fn, *args, **kwargs) + + +def is_err_and_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches an async + predicate.""" + return result.is_err_and_async(fn, *args, **kwargs) + + +def is_err_or[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches a + predicate.""" + return result.is_err_or(fn, *args, **kwargs) + + +def is_err_or_async[T, E, **P]( + result: r.FutureResult[T, E], + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches an async + predicate.""" + return result.is_err_or_async(fn, *args, **kwargs) diff --git a/wird/result.py b/wird/result.py new file mode 100644 index 0000000..131ee7c --- /dev/null +++ b/wird/result.py @@ -0,0 +1,452 @@ +from typing import Any, Awaitable, Callable, Concatenate, Type, overload + +from . import _future as f, _result as r + +__all__ = ( + "and_", + "and_then", + "and_then_async", + "inspect", + "inspect_async", + "inspect_err", + "inspect_err_async", + "is_err", + "is_err_and", + "is_err_and_async", + "is_err_or", + "is_err_or_async", + "is_ok", + "is_ok_and", + "is_ok_and_async", + "is_ok_or", + "is_ok_or_async", + "map", + "map_async", + "map_err", + "map_err_async", + "or_", + "or_else", + "or_else_async", + "unwrap", + "unwrap_err", + "unwrap_err_or", + "unwrap_err_or_else", + "unwrap_err_or_else_async", + "unwrap_or", + "unwrap_or_else", + "unwrap_or_else_async", +) + + +@overload +def unwrap[T, E, R]( + res: r.Result[T, E], + *, + as_type: Type[R], + on_err: str = "expected Ok, got Err", +) -> R: + """Returns the contained Ok value casted to passed type. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + +@overload +def unwrap[T, E](res: r.Result[T, E], *, on_err: str = "expected Ok, got Err") -> T: + """Returns the contained Ok value. + + Because this function may raise ErrUnwrapError, its use is generally + discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_or, unwrap_or_else. + """ + ... + + +def unwrap[T, E](res: r.Result[T, E], **kwargs) -> Any: + return res.unwrap(**kwargs) + + +def unwrap_or[T, E](res: r.Result[T, E], /, other: T) -> T: + """Returns the contained Ok value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_or_else, which is + lazily evaluated. + """ + return res.unwrap_or(other) + + +def unwrap_or_else[T, E, **P]( + res: r.Result[T, E], + fn: Callable[P, T], + *args: P.args, + **kwargs: P.kwargs, +) -> T: + """Returns the contained Ok value or computes it from a closure.""" + return res.unwrap_or_else(fn, *args, **kwargs) + + +def unwrap_or_else_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[P, Awaitable[T]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[T]: + """Returns the contained Ok value or computes it from an async closure.""" + return res.unwrap_or_else_async(fn, *args, **kwargs) + + +@overload +def unwrap_err[T, E, R]( + res: r.Result[T, E], + *, + as_type: Type[R], + on_ok: str = "expected Err, got Ok", +) -> R: + """Returns the contained Err value casted to passed type. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + + No actual casting is performed, type change affects only type checkers. + """ + ... + + +@overload +def unwrap_err[T, E](res: r.Result[T, E], *, on_ok: str = "expected Err, got Ok") -> E: + """Returns the contained Ok value. + + Because this function may raise OkUnwrapError, its use is generally discouraged. + + Instead, prefer to use pattern matching and handle the Err case explicitly, or + call unwrap_err_or, unwrap_err_or_else. + """ + ... + + +def unwrap_err[T, E](res: r.Result[T, E], **kwargs) -> Any: + return res.unwrap_err(**kwargs) + + +def unwrap_err_or[T, E](res: r.Result[T, E], /, other: E) -> E: + """Returns the contained Err value or a provided other. + + Arguments passed to unwrap_or are eagerly evaluated; if you are passing the + result of a function call, it is recommended to use unwrap_err_or_else, which is + lazily evaluated. + """ + return res.unwrap_err_or(other) + + +def unwrap_err_or_else[T, E, **P]( + res: r.Result[T, E], + fn: Callable[P, E], + *args: P.args, + **kwargs: P.kwargs, +) -> E: + """Returns the contained Err value or computes it from a closure.""" + return res.unwrap_err_or_else(fn, *args, **kwargs) + + +def unwrap_err_or_else_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[P, Awaitable[E]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[E]: + """Returns the contained Err value or computes it from an async closure.""" + return res.unwrap_err_or_else_async(fn, *args, **kwargs) + + +def map[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], R], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[R, E]: + """Maps a Result[T, E] to Result[R, E] by applying a function to a contained Ok + value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return res.map(fn, *args, **kwargs) + + +def map_async[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Maps a Result[T, E] to FutureResult[R, E] by applying an async function to a + contained Ok value, leaving an Err value untouched. + + This function can be used to compose the results of two functions. + """ + return res.map_async(fn, *args, **kwargs) + + +def inspect[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Any], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[T, E]: + """Calls a function with a reference to the contained value if Ok. + + Returns the original res. + """ + return res.inspect(fn, *args, **kwargs) + + +def inspect_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as Futureres. + """ + return res.inspect_async(fn, *args, **kwargs) + + +def map_err[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], R], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[T, R]: + """Maps a Result[T, E] to Result[T, R] by applying a function to a contained Err + value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return res.map_err(fn, *args, **kwargs) + + +def map_err_async[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Awaitable[R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Maps a Result[T, E] to FutureResult[T, R] by applying am async function to a + contained Err value, leaving an Ok value untouched. + + This function can be used to pass through a successful result while handling an + error. + """ + return res.map_err_async(fn, *args, **kwargs) + + +def inspect_err[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Any], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[T, E]: + """Calls a function with a reference to the contained value if Err. + + Returns the original res. + """ + return res.inspect_err(fn, *args, **kwargs) + + +def inspect_err_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Awaitable[Any]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, E]: + """Calls an async function with a reference to the contained value if Ok. + + Returns the original result as Futureres. + """ + return res.inspect_err_async(fn, *args, **kwargs) + + +def and_[T, E, R](res: r.Result[T, E], other: r.Result[R, E]) -> r.Result[R, E]: + """Returns other if the result is Ok, otherwise returns the Err value of res: r.Result[T, E]. + + Arguments passed to and are eagerly evaluated; if you are passing the result of + a function call, it is recommended to use and_then, which is lazily evaluated. + """ + return res.and_(other) + + +def and_then[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], r.Result[R, E]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[R, E]: + """Calls a function if the result is Ok, otherwise returns the Err value of + res: r.Result[T, E]. + + This function can be used for control flow based on Result values. + """ + return res.and_then(fn, *args, **kwargs) + + +def and_then_async[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Awaitable[r.Result[R, E]]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[R, E]: + """Calls an async function if the result is Ok, otherwise returns the Err value + of res: r.Result[T, E]. + + This function can be used for control flow based on Result values. + """ + return res.and_then_async(fn, *args, **kwargs) + + +def or_[T, E, R](res: r.Result[T, E], other: r.Result[T, R]) -> r.Result[T, R]: + """Returns other if the result is Err, otherwise returns the Ok value of res: r.Result[T, E]. + + Arguments passed to or are eagerly evaluated; if you are passing the result of a + function call, it is recommended to use or_else, which is lazily evaluated. + """ + return res.or_(other) + + +def or_else[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], r.Result[T, R]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.Result[T, R]: + """Calls a function if the result is Err, otherwise returns the Ok value of + res: r.Result[T, E]. + + This function can be used for control flow based on result values. + """ + return res.or_else(fn, *args, **kwargs) + + +def or_else_async[T, E, **P, R]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Awaitable[r.Result[T, R]]], + *args: P.args, + **kwargs: P.kwargs, +) -> r.FutureResult[T, R]: + """Calls an async function if the result is Err, otherwise returns the Ok value + of res: r.Result[T, E]. + + This function can be used for control flow based on result values. + """ + return res.or_else_async(fn, *args, **kwargs) + + +def is_ok[T, E](res: r.Result[T, E]) -> bool: + """Returns True if the result is Ok.""" + return res.is_ok() + + +def is_ok_and[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> bool: + """Returns True if the result is Ok and the value inside of it matches a + predicate.""" + return res.is_ok_and(fn, *args, **kwargs) + + +def is_ok_and_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok and the value inside of it matches an async + predicate.""" + return res.is_ok_and_async(fn, *args, **kwargs) + + +def is_ok_or[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> bool: + """Returns True if the result is Ok or the value inside of Err matches a + predicate.""" + return res.is_ok_or(fn, *args, **kwargs) + + +def is_ok_or_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Ok or the value inside of Err matches an async + predicate.""" + return res.is_ok_or_async(fn, *args, **kwargs) + + +def is_err[T, E](res: r.Result[T, E]) -> bool: + """Returns True if the result is Err.""" + return res.is_err() + + +def is_err_and[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> bool: + """Returns True if the result is Err and the value inside of it matches a + predicate.""" + return res.is_err_and(fn, *args, **kwargs) + + +def is_err_and_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[E, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err and the value inside of it matches an async + predicate.""" + return res.is_err_and_async(fn, *args, **kwargs) + + +def is_err_or[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], bool], + *args: P.args, + **kwargs: P.kwargs, +) -> bool: + """Returns True if the result is Err or the value inside of Ok matches a + predicate.""" + return res.is_err_or(fn, *args, **kwargs) + + +def is_err_or_async[T, E, **P]( + res: r.Result[T, E], + fn: Callable[Concatenate[T, P], Awaitable[bool]], + *args: P.args, + **kwargs: P.kwargs, +) -> f.Future[bool]: + """Returns True if the result is Err or the value inside of Ok matches an async + predicate.""" + return res.is_err_or_async(fn, *args, **kwargs)