trcks.oop.AwaitableWrapper

class documentation

class AwaitableWrapper(_AwaitableWrapper[_T_co]): (source)

Constructor: AwaitableWrapper(core)

Typesafe and immutable wrapper for collections.abc.Awaitable objects.

The wrapped collections.abc.Awaitable can be accessed via the attribute AwaitableWrapper.core. The AwaitableWrapper.map* methods allow method chaining.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> async def read_from_disk() -> str:
...     await asyncio.sleep(0.001)
...     input_ = "Hello, world!"
...     print(f"Read '{input_}' from disk.")
...     return input_
...
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> async def write_to_disk(output: str) -> None:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{output}' to disk.")
...
>>> async def main() -> None:
...     awaitable_str = read_from_disk()
...     return await (
...         AwaitableWrapper
...         .construct_from_awaitable(awaitable_str)
...         .map(transform)
...         .map_to_awaitable(write_to_disk)
...         .core
...     )
...
>>> asyncio.run(main())
Read 'Hello, world!' from disk.
Wrote 'Length: 13' to disk.

Static Method	`construct`	Construct and wrap an `collections.abc.Awaitable` object from a value.
Static Method	`construct_from_awaitable`	Alias for the default constructor.
Method	`map`	Apply a sync. function to the wrapped `collections.abc.Awaitable` object.
Method	`map_to_awaitable`	Apply an async. function to the wrapped `collections.abc.Awaitable` object.
Method	`map_to_awaitable_result`	Apply an asynchronous function with return type `trcks.Result`.
Method	`map_to_result`	Apply a sync. function with return type `trcks.Result`.

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(value: _T) -> AwaitableWrapper[_T]: (source) ¶

Construct and wrap an collections.abc.Awaitable object from a value.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> awaitable_wrapper = AwaitableWrapper.construct("Hello, world!")
>>> awaitable_wrapper
AwaitableWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_wrapper.core_as_coroutine)
'Hello, world!'

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

Returns
`AwaitableWrapper[_T]`	A new `AwaitableWrapper` instance with the wrapped `collections.abc.Awaitable` object.

@staticmethod

def construct_from_awaitable(awtbl: Awaitable[_T]) -> AwaitableWrapper[_T]: (source) ¶

Alias for the default constructor.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> async def read_from_disk() -> str:
...     return await asyncio.sleep(0.001, result="Hello, world!")
...
>>> awaitable_str = read_from_disk()
>>> awaitable_wrapper = AwaitableWrapper.construct_from_awaitable(
...     awaitable_str
... )
>>> awaitable_wrapper
AwaitableWrapper(core=<coroutine object read_from_disk at 0x...>)
>>> asyncio.run(awaitable_wrapper.core_as_coroutine)
'Hello, world!'

Parameters
awtbl:`Awaitable[_T]`	The `collections.abc.Awaitable` to be wrapped.

Returns
`AwaitableWrapper[_T]`	A new `AwaitableWrapper` instance with the wrapped `collections.abc.Awaitable` object.

def map(self, f: Callable[[_T_co], _T]) -> AwaitableWrapper[_T]: (source) ¶

Apply a sync. function to the wrapped collections.abc.Awaitable object.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> awaitable_wrapper = (
...     AwaitableWrapper
...     .construct("Hello, world!")
...     .map(transform)
... )
>>> awaitable_wrapper
AwaitableWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_wrapper.core_as_coroutine)
'Length: 13'

Parameters
f:`Callable[[_T_co], _T]`	The synchronous function to be applied.

Returns
`AwaitableWrapper[_T]`	A new `AwaitableWrapper` instance with the result of the function application.

def map_to_awaitable(self, f: Callable[[_T_co], Awaitable[_T]]) -> AwaitableWrapper[_T]: (source) ¶

Apply an async. function to the wrapped collections.abc.Awaitable object.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> async def write_to_disk(output: str) -> None:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{output}' to disk.")
...
>>> awaitable_wrapper = (
...     AwaitableWrapper
...     .construct("Hello, world!")
...     .map_to_awaitable(write_to_disk)
... )
>>> awaitable_wrapper
AwaitableWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_wrapper.core_as_coroutine)
Wrote 'Hello, world!' to disk.

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

Returns
`AwaitableWrapper[_T]`	A new `AwaitableWrapper` instance with the result of the function application.

def map_to_awaitable_result(self, f: Callable[[_T_co], AwaitableResult[_F, _S]]) -> AwaitableResultWrapper[_F, _S]: (source) ¶

Apply an asynchronous function with return type trcks.Result.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.oop import AwaitableWrapper
>>> async def slowly_assert_non_negative(
...     x: float,
... ) -> Result[str, float]:
...     await asyncio.sleep(0.001)
...     if x < 0:
...         return "failure", "negative value"
...     return "success", x
...
>>> awaitable_result_wrapper = (
...     AwaitableWrapper
...     .construct(42.0)
...     .map_to_awaitable_result(slowly_assert_non_negative)
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('success', 42.0)

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

Returns
`AwaitableResultWrapper[_F, _S]`	An `AwaitableResultWrapper` instance with the result of the function application.

def map_to_result(self, f: Callable[[_T_co], Result[_F, _S]]) -> AwaitableResultWrapper[_F, _S]: (source) ¶

Apply a sync. function with return type trcks.Result.

Example

>>> import asyncio
>>> from trcks.oop import AwaitableWrapper
>>> awaitable_result_wrapper = (
...     AwaitableWrapper
...     .construct(-1)
...     .map_to_result(
...         lambda x: ("success", x)
...         if x >= 0
...         else ("failure", "negative value")
...     )
... )
>>> awaitable_result_wrapper
AwaitableResultWrapper(core=<coroutine object ...>)
>>> asyncio.run(awaitable_result_wrapper.core_as_coroutine)
('failure', 'negative value')

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

Returns
`AwaitableResultWrapper[_F, _S]`	An `AwaitableResultWrapper` instance with the result of the function application.