trcks.oop.ResultWrapper

class documentation

class ResultWrapper(_Wrapper[Result[_F_default_co, _S_default_co]]): (source)

Constructor: ResultWrapper(core)

Typesafe and immutable wrapper for trcks.Result objects.

The wrapped object can be accessed via the attribute ResultWrapper.core. The ResultWrapper.map* methods allow method chaining.

Example

>>> import math
>>> from trcks.oop import ResultWrapper
>>> result_wrapper = (
...     ResultWrapper
...     .construct_success(-5.0)
...     .map_success_to_result(
...         lambda x: ("success", x)
...         if x >= 0
...         else ("failure", "negative value")
...     )
...     .map_success(math.sqrt)
... )
>>> result_wrapper
ResultWrapper(core=('failure', 'negative value'))
>>> result_wrapper.core
('failure', 'negative value')

Static Method	`construct_failure`	Construct and wrap a `trcks.Failure` object from a value.
Static Method	`construct_from_result`	Wrap a `trcks.Result` object.
Static Method	`construct_success`	Construct and wrap a `trcks.Success` object from a value.
Method	`map_failure`	Apply sync. func. to the wrapped `trcks.Result` object if it is a failure.
Method	`map_failure_to_awaitable`	Apply async. func. to the wrapped `trcks.Result` object if it is a failure.
Method	`map_failure_to_awaitable_result`	Apply async. `trcks.Result` func. to wrapped `trcks.Result` obj. if failure.
Method	`map_failure_to_result`	Apply sync. `trcks.Result` func. to wrapped `trcks.Result` obj. if failure.
Method	`map_success`	Apply sync. func. to the wrapped `trcks.Result` object if it is a success.
Method	`map_success_to_awaitable`	Apply async. func. to the wrapped `trcks.Result` object if it is a success.
Method	`map_success_to_awaitable_result`	Apply async. `trcks.Result` func. to wrapped `trcks.Result` obj. if success.
Method	`map_success_to_result`	Apply sync. `trcks.Result` func. to wrapped `trcks.Result` obj. if success.
Property	`track`	First element of the attribute `ResultWrapper.core`.
Property	`value`	Second element of the attribute `ResultWrapper.core`.

Inherited from _Wrapper:

Method	`__init__`	Construct wrapper.
Method	`__repr__`	Return a string representation of the wrapper.
Property	`core`	The wrapped object.
Instance Variable	`_core`	Undocumented

@staticmethod

def construct_failure(value: _F) -> ResultWrapper[_F, Never]: (source) ¶

Construct and wrap a trcks.Failure object from a value.

Example

>>> ResultWrapper.construct_failure(42)
ResultWrapper(core=('failure', 42))

Parameters
value:`_F`	The value to be wrapped.

Returns
`ResultWrapper[_F, Never]`	A new `ResultWrapper` instance with the wrapped `trcks.Failure` object.

@staticmethod

def construct_from_result(rslt: Result[_F_default, _S_default]) -> ResultWrapper[_F_default, _S_default]: (source) ¶

Wrap a trcks.Result object.

Example

>>> ResultWrapper.construct_from_result(("success", 0.0))
ResultWrapper(core=('success', 0.0))

Parameters
rslt:`Result[_F_default, _S_default]`	The `trcks.Result` object to be wrapped.

Returns
`ResultWrapper[_F_default, _S_default]`	A new `ResultWrapper` instance with the wrapped `trcks.Result` object.

@staticmethod

def construct_success(value: _S) -> ResultWrapper[Never, _S]: (source) ¶

Construct and wrap a trcks.Success object from a value.

Example

>>> ResultWrapper.construct_success(42)
ResultWrapper(core=('success', 42))

Parameters
value:`_S`	The value to be wrapped.

Returns
`ResultWrapper[Never, _S]`	A new `ResultWrapper` instance with the wrapped `trcks.Success` object.

def map_failure(self, f: Callable[[_F_default_co], _F]) -> ResultWrapper[_F, _S_default_co]: (source) ¶

Apply sync. func. to the wrapped trcks.Result object if it is a failure.

Example

>>> ResultWrapper.construct_failure("negative value").map_failure(
...     lambda s: f"Prefix: {s}"
... )
ResultWrapper(core=('failure', 'Prefix: negative value'))
>>>
>>> ResultWrapper.construct_success(25.0).map_failure(
...     lambda s: f"Prefix: {s}"
... )
ResultWrapper(core=('success', 25.0))

Parameters
f:`Callable[[_F_default_co], _F]`	The synchronous function to be applied.

Returns

ResultWrapper[_F, _S_default_co]

A new ResultWrapper instance with

the result of the function application if the original trcks.Result is a failure, or
the original trcks.Result object if it is a success.

def map_failure_to_awaitable(self, f: Callable[[_F_default_co], Awaitable[_F]]) -> AwaitableResultWrapper[_F, _S_default_co]: (source) ¶

Apply async. func. to the wrapped trcks.Result object if it is a failure.

Example

>>> import asyncio
>>> from trcks.oop import ResultWrapper
>>> async def add_prefix_slowly(s: str) -> str:
...     await asyncio.sleep(0.001)
...     return f"Prefix: {s}"
...
>>> awaitable_result_wrapper_1 = (
...     ResultWrapper
...     .construct_failure("not found")
...     .map_failure_to_awaitable(add_prefix_slowly)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('failure', 'Prefix: not found')
>>>
>>> awaitable_result_wrapper_2 = (
...     ResultWrapper
...     .construct_success(42)
...     .map_failure_to_awaitable(add_prefix_slowly)
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('success', 42)

Parameters
f:`Callable[[_F_default_co], Awaitable[_F]]`	The asynchronous function to be applied.

Returns

AwaitableResultWrapper[_F, _S_default_co]

An AwaitableResultWrapper instance with

the result of the function application if the original trcks.Result is a failure, or
the original trcks.Result object if it is a success.

def map_failure_to_awaitable_result(self, f: Callable[[_F_default_co], AwaitableResult[_F, _S]]) -> AwaitableResultWrapper[_F, _S_default_co | _S]: (source) ¶

Apply async. trcks.Result func. to wrapped trcks.Result obj. if failure.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.oop import ResultWrapper
>>> async def slowly_replace_not_found(s: str) -> Result[str, float]:
...     await asyncio.sleep(0.001)
...     if s == "not found":
...         return "success", 0.0
...     return "failure", s
...
>>> awaitable_result_wrapper_1 = (
...     ResultWrapper
...     .construct_failure("not found")
...     .map_failure_to_awaitable_result(slowly_replace_not_found)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('success', 0.0)
>>>
>>> awaitable_result_wrapper_2 = (
...     ResultWrapper
...     .construct_failure("other failure")
...     .map_failure_to_awaitable_result(slowly_replace_not_found)
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('failure', 'other failure')
>>>
>>> awaitable_result_wrapper_3 = (
...     ResultWrapper
...     .construct_success(25.0)
...     .map_failure_to_awaitable_result(slowly_replace_not_found)
... )
>>> awaitable_result_wrapper_3
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_3.core_as_coroutine)
('success', 25.0)

Parameters
f:`Callable[[_F_default_co], AwaitableResult[_F, _S]]`	The asynchronous function to be applied.

Returns

AwaitableResultWrapper[_F, _S_default_co | _S]

An AwaitableResultWrapper instance with

the result of the function application if the original trcks.Result is a failure, or
the original trcks.Result object if it is a success.

def map_failure_to_result(self, f: Callable[[_F_default_co], Result[_F, _S]]) -> ResultWrapper[_F, _S_default_co | _S]: (source) ¶

Apply sync. trcks.Result func. to wrapped trcks.Result obj. if failure.

Example

>>> from trcks import Result
>>> from trcks.oop import ResultWrapper
>>> def replace_not_found_by_default_value(s: str) -> Result[str, float]:
...     if s == "not found":
...         return "success", 0.0
...     return "failure", s
...
>>> ResultWrapper.construct_failure(
...     "not found"
... ).map_failure_to_result(
...     replace_not_found_by_default_value
... )
ResultWrapper(core=('success', 0.0))
>>>
>>> ResultWrapper.construct_failure(
...     "other failure"
... ).map_failure_to_result(
...     replace_not_found_by_default_value
... )
ResultWrapper(core=('failure', 'other failure'))
>>>
>>> ResultWrapper.construct_success(
...     25.0
... ).map_failure_to_result(
...     replace_not_found_by_default_value
... )
ResultWrapper(core=('success', 25.0))

Parameters
f:`Callable[[_F_default_co], Result[_F, _S]]`	The synchronous function to be applied.

Returns

ResultWrapper[_F, _S_default_co | _S]

A new ResultWrapper instance with

the result of the function application if the original trcks.Result is a failure, or
the original trcks.Result object if it is a success.

def map_success(self, f: Callable[[_S_default_co], _S]) -> ResultWrapper[_F_default_co, _S]: (source) ¶

Apply sync. func. to the wrapped trcks.Result object if it is a success.

Example

>>> ResultWrapper.construct_failure("not found").map_success(lambda n: n+1)
ResultWrapper(core=('failure', 'not found'))
>>>
>>> ResultWrapper.construct_success(42).map_success(lambda n: n+1)
ResultWrapper(core=('success', 43))

Parameters
f:`Callable[[_S_default_co], _S]`	The synchronous function to be applied.

Returns

ResultWrapper[_F_default_co, _S]

A new ResultWrapper instance with

the original trcks.Result object if it is a failure, or
the result of the function application if the original trcks.Result is a success.

def map_success_to_awaitable(self, f: Callable[[_S_default_co], Awaitable[_S]]) -> AwaitableResultWrapper[_F_default_co, _S]: (source) ¶

Apply async. func. to the wrapped trcks.Result object if it is a success.

Example

>>> import asyncio
>>> from trcks.oop import ResultWrapper
>>> async def increment_slowly(n: int) -> int:
...     await asyncio.sleep(0.001)
...     return n + 1
...
>>> awaitable_result_wrapper_1 = (
...     ResultWrapper
...     .construct_failure("not found")
...     .map_success_to_awaitable(increment_slowly)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('failure', 'not found')
>>>
>>> awaitable_result_wrapper_2 = (
...     ResultWrapper
...     .construct_success(42)
...     .map_success_to_awaitable(increment_slowly)
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('success', 43)

Parameters
f:`Callable[[_S_default_co], Awaitable[_S]]`	The asynchronous function to be applied.

Returns

AwaitableResultWrapper[_F_default_co, _S]

An AwaitableResultWrapper instance with

the original trcks.Result object if it is a failure, or
the result of the function application if the original trcks.Result is a success.

def map_success_to_awaitable_result(self, f: Callable[[_S_default_co], AwaitableResult[_F, _S]]) -> AwaitableResultWrapper[_F_default_co | _F, _S]: (source) ¶

Apply async. trcks.Result func. to wrapped trcks.Result obj. if success.

Example

>>> import asyncio
>>> import math
>>> from trcks import Result
>>> from trcks.oop import ResultWrapper
>>> async def get_square_root_slowly(x: float) -> Result[str, float]:
...     await asyncio.sleep(0.001)
...     if x < 0:
...         return "failure", "negative value"
...     return "success", math.sqrt(x)
...
>>> awaitable_result_wrapper_1 = (
...     ResultWrapper
...     .construct_failure("not found")
...     .map_success_to_awaitable_result(get_square_root_slowly)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('failure', 'not found')
>>>
>>> awaitable_result_wrapper_2 = (
...     ResultWrapper
...     .construct_success(-25.0)
...     .map_success_to_awaitable_result(get_square_root_slowly)
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('failure', 'negative value')
>>>
>>> awaitable_result_wrapper_3 = (
...     ResultWrapper
...     .construct_success(25.0)
...     .map_success_to_awaitable_result(get_square_root_slowly)
... )
>>> awaitable_result_wrapper_3
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_3.core_as_coroutine)
('success', 5.0)

Parameters
f:`Callable[[_S_default_co], AwaitableResult[_F, _S]]`	The asynchronous function to be applied.

Returns

AwaitableResultWrapper[_F_default_co | _F, _S]

An AwaitableResultWrapper instance with

the original trcks.Result object if it is a failure, or
the result of the function application if the original trcks.Result is a success.

def map_success_to_result(self, f: Callable[[_S_default_co], Result[_F, _S]]) -> ResultWrapper[_F_default_co | _F, _S]: (source) ¶

Apply sync. trcks.Result func. to wrapped trcks.Result obj. if success.

Example

>>> import math
>>> from trcks import Result
>>> from trcks.oop import ResultWrapper
>>> def get_square_root(x: float) -> Result[str, float]:
...     if x < 0:
...         return "failure", "negative value"
...     return "success", math.sqrt(x)
...
>>> ResultWrapper.construct_failure("not found").map_success_to_result(
...     get_square_root
... )
ResultWrapper(core=('failure', 'not found'))
>>>
>>> ResultWrapper.construct_success(-25.0).map_success_to_result(
...     get_square_root
... )
ResultWrapper(core=('failure', 'negative value'))
>>>
>>> ResultWrapper.construct_success(25.0).map_success_to_result(
...     get_square_root
... )
ResultWrapper(core=('success', 5.0))

Parameters
f:`Callable[[_S_default_co], Result[_F, _S]]`	The synchronous function to be applied.

Returns

ResultWrapper[_F_default_co | _F, _S]

A new ResultWrapper instance with

the original trcks.Result object if it is a failure, or
the result of the function application if the original trcks.Result is a success.

@property

track: Literal['failure', 'success'] = (source) ¶

First element of the attribute ResultWrapper.core.

Example

>>> ResultWrapper(core=('failure', 42)).track
'failure'

@property

value: _F_default_co | _S_default_co = (source) ¶

Second element of the attribute ResultWrapper.core.

Example

>>> ResultWrapper(core=('failure', 42)).value
42