trcks.oop.AwaitableResultWrapper

class documentation

class AwaitableResultWrapper(_AwaitableWrapper[Result[_F_default_co, _S_default_co]]): (source)

Constructor: AwaitableResultWrapper(core)

Typesafe and immutable wrapper for trcks.AwaitableResult objects.

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

Example

>>> import asyncio
>>> import math
>>> from trcks import Result
>>> from trcks.oop import AwaitableResultWrapper
>>> async def read_from_disk() -> Result[str, float]:
...     await asyncio.sleep(0.001)
...     return "failure", "not found"
...
>>> def get_square_root(x: float) -> Result[str, float]:
...     if x < 0:
...         return "failure", "negative value"
...     return "success", math.sqrt(x)
...
>>> async def write_to_disk(output: float) -> None:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{output}' to disk.")
...
>>> async def main() -> Result[str, None]:
...     awaitable_result = read_from_disk()
...     return await (
...         AwaitableResultWrapper
...         .construct_from_awaitable_result(awaitable_result)
...         .map_success_to_result(get_square_root)
...         .map_success_to_awaitable(write_to_disk)
...         .core
...     )
...
>>> asyncio.run(main())
('failure', 'not found')

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

Inherited from _AwaitableWrapper:

Property core_as_coroutine The wrapped collections.abc.Awaitable object transformed into a coroutine.

Inherited from _Wrapper (via _AwaitableWrapper):

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) -> AwaitableResultWrapper[_F, Never]: (source) ¶

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

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> awaitable_result_wrapper = (
...     AwaitableResultWrapper.construct_failure("not found")
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('failure', 'not found')

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

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

@staticmethod

def construct_failure_from_awaitable(awtbl: Awaitable[_F]) -> AwaitableResultWrapper[_F, Never]: (source) ¶

Construct and wrap an awaitable trcks.Failure from an awaitable value.

Example

>>> import asyncio
>>> from http import HTTPStatus
>>> from trcks.oop import AwaitableResultWrapper
>>> async def get_status() -> HTTPStatus:
...     await asyncio.sleep(0.001)
...     return HTTPStatus.NOT_FOUND
...
>>> awaitable_status = get_status()
>>> awaitable_result_wrapper = (
...     AwaitableResultWrapper
...     .construct_failure_from_awaitable(awaitable_status)
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('failure', <HTTPStatus.NOT_FOUND: 404>)

Parameters
awtbl:`Awaitable[_F]`	The awaitable value to be wrapped.

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

@staticmethod

def construct_from_awaitable_result(a_rslt: AwaitableResult[_F, _S]) -> AwaitableResultWrapper[_F, _S]: (source) ¶

Wrap an awaitable trcks.Result object.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.oop import AwaitableResultWrapper
>>> async def read_from_disk() -> Result[str, str]:
...     await asyncio.sleep(0.001)
...     return "failure", "file not found"
...
>>> awaitable_result = read_from_disk()
>>> awaitable_wrapper = (
...     AwaitableResultWrapper
...     .construct_from_awaitable_result(awaitable_result)
... )
>>> awaitable_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_wrapper.core_as_coroutine)
('failure', 'file not found')

Parameters
a_rslt:`AwaitableResult[_F, _S]`	The awaitable `trcks.Result` object to be wrapped.

Returns
`AwaitableResultWrapper[_F, _S]`	A new `AwaitableResultWrapper` instance with the wrapped `trcks.AwaitableResult` object.

@staticmethod

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

Construct and wrap an awaitbl. trcks.Result obj. from a trcks.Result obj.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> awaitable_result_wrapper = (
...     AwaitableResultWrapper
...     .construct_from_result(("failure", "not found"))
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('failure', 'not found')

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

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

@staticmethod

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

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

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> awaitable_result_wrapper = AwaitableResultWrapper.construct_success(42)
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('success', 42)

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

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

@staticmethod

def construct_success_from_awaitable(awtbl: Awaitable[_S]) -> AwaitableResultWrapper[Never, _S]: (source) ¶

Construct and wrap an awaitable trcks.Success from an awaitable value.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> async def read_from_disk() -> str:
...     await asyncio.sleep(0.001)
...     return "Hello, world!"
...
>>> awaitable_str = read_from_disk()
>>> awaitable_result_wrapper = (
...     AwaitableResultWrapper
...     .construct_success_from_awaitable(awaitable_str)
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('success', 'Hello, world!')

Parameters
awtbl:`Awaitable[_S]`	The awaitable value to be wrapped.

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

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

Apply sync. func. to wrapped trcks.AwaitableResult obj. if it is a failure.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> awaitable_result_wrapper_1 = (
...     AwaitableResultWrapper
...     .construct_failure("negative value")
...     .map_failure(lambda s: f"Prefix: {s}")
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('failure', 'Prefix: negative value')
>>>
>>> awaitable_result_wrapper_2 = (
...     AwaitableResultWrapper
...     .construct_success(25.0)
...     .map_failure(lambda s: f"Prefix: {s}")
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('success', 25.0)

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

Returns

AwaitableResultWrapper[_F, _S_default_co]

A new AwaitableResultWrapper instance with

the result of the function application if the original trcks.AwaitableResult is a failure, or
the original trcks.AwaitableResult 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 wrapped trcks.AwaitableResult obj. if failure.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> async def add_prefix_slowly(s: str) -> str:
...     await asyncio.sleep(0.001)
...     return f"Prefix: {s}"
...
>>> awaitable_result_wrapper_1 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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]

A new AwaitableResultWrapper instance with

the result of the function application if the original trcks.AwaitableResult is a failure, or
the original trcks.AwaitableResult 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 trcks.AwaitableResult if failure.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.oop import AwaitableResultWrapper
>>> 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 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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]

A new AwaitableResultWrapper instance with

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

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

Apply sync. trcks.Result func. to trcks.AwaitableResult if failure.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.oop import AwaitableResultWrapper
>>> def replace_not_found_by_default_value(s: str) -> Result[str, float]:
...     if s == "not found":
...         return "success", 0.0
...     return "failure", s
...
>>> awaitable_result_wrapper_1 = (
...     AwaitableResultWrapper
...     .construct_failure("not found")
...     .map_failure_to_result(replace_not_found_by_default_value)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('success', 0.0)
>>>
>>> awaitable_result_wrapper_2 = (
...     AwaitableResultWrapper
...     .construct_failure("other failure")
...     .map_failure_to_result(replace_not_found_by_default_value)
... )
>>> awaitable_result_wrapper_2
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_2.core_as_coroutine)
('failure', 'other failure')
>>>
>>> awaitable_result_wrapper_3 = (
...     AwaitableResultWrapper
...     .construct_success(25.0)
...     .map_failure_to_result(replace_not_found_by_default_value)
... )
>>> 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], Result[_F, _S]]`	The synchronous function to be applied.

Returns

AwaitableResultWrapper[_F, _S_default_co | _S]

A new AwaitableResultWrapper instance with

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

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

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

Example

>>> import asyncio
>>> import math
>>> from trcks.oop import AwaitableResultWrapper
>>> awaitable_result_wrapper_1 = (
...     AwaitableResultWrapper
...     .construct_failure("not found")
...     .map_success(lambda n: n + 1)
... )
>>> awaitable_result_wrapper_1
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper_1.core_as_coroutine)
('failure', 'not found')
>>>
>>> awaitable_result_wrapper_2 = (
...     AwaitableResultWrapper
...     .construct_success(42)
...     .map_success(lambda n: n + 1)
... )
>>> 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], _S]`	The synchronous function to be applied.

Returns

AwaitableResultWrapper[_F_default_co, _S]

A new AwaitableResultWrapper instance with

the original trcks.AwaitableResult object if it is a failure, or
the result of the function application if the original trcks.AwaitableResult 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 wrapped trcks.AwaitableResult obj. if success.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> async def increment_slowly(n: int) -> int:
...     await asyncio.sleep(0.001)
...     return n + 1
...
>>> awaitable_result_wrapper_1 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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]

A new AwaitableResultWrapper instance with

the original trcks.AwaitableResult object if it is a failure, or
the result of the function application if the original trcks.AwaitableResult 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 trcks.AwaitableResult if success.

Example

>>> import asyncio
>>> import math
>>> from trcks import Result
>>> from trcks.oop import AwaitableResultWrapper
>>> 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 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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 = (
...     AwaitableResultWrapper
...     .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]

A new AwaitableResultWrapper instance with

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

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

Apply sync. trcks.Result func. to trcks.AwaitableResult if success.

Example

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

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

Returns

AwaitableResultWrapper[_F_default_co | _F, _S]

A new AwaitableResultWrapper instance with

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

@property

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

First element of the awaited attribute AwaitableResultWrapper.core.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> track_coroutine = AwaitableResultWrapper.construct_failure(42).track
>>> track_coroutine
<coroutine object ...>
>>> asyncio.run(track_coroutine)
'failure'

@property

value: _F_default_co | _S_default_co = (source) ¶

Second element of the awaited attribute AwaitableResultWrapper.core.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableResultWrapper
>>> value_coroutine = AwaitableResultWrapper.construct_failure(42).value
>>> value_coroutine
<coroutine object ...>
>>> asyncio.run(value_coroutine)
42