trcks.fp.monads.awaitable

module documentation

(source)

Monadic functions for trcks.AwaitableResult.

Provides utilities for functional composition of trcks.AwaitableResult-returning functions.

Example

>>> import asyncio
>>> import math
>>> from trcks import Result
>>> from trcks.fp.composition import pipe
>>> from trcks.fp.monads import awaitable_result as ar
>>> 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 pipe(
...         (
...             awaitable_result,
...             ar.map_success_to_result(get_square_root),
...             ar.map_success_to_awaitable(write_to_disk),
...         )
...     )
...
>>> asyncio.run(main())
('failure', 'not found')

Function	`construct_failure`	Create an `AwaitableFailure` object from a value.
Function	`construct_failure_from_awaitable`	Create an `AwaitableFailure` object from an `Awaitable` object.
Function	`construct_from_result`	Create an `AwaitableResult` object from a `Result` object.
Function	`construct_success`	Create an `AwaitableSuccess` object from a value.
Function	`construct_success_from_awaitable`	Create an `AwaitableSuccess` object from an `Awaitable` object.
Function	`map_failure`	Create function that maps `AwaitableFailure` to `AwaitableFailure` values.
Function	`map_failure_to_awaitable`	Create function that maps `AwaitableFailure` to `AwaitableFailure` values.
Function	`map_failure_to_awaitable_result`	Create function that maps `AwaitableFailure` values to `AwaitableResult` values.
Function	`map_failure_to_result`	Create function that maps `AwaitableFailure` values to `AwaitableResult` values.
Function	`map_success`	Creates function that maps `AwaitableSuccess` to `AwaitableSuccess` values.
Function	`map_success_to_awaitable`	Creates function that maps `AwaitableSuccess` to `AwaitableSuccess` values.
Function	`map_success_to_awaitable_result`	Create function that maps `AwaitableSuccess` values to `AwaitableResult` values.
Function	`map_success_to_result`	Creates function that maps `AwaitableSuccess` values to `AwaitableResult` values.
Async Function	`to_coroutine_result`	Turn an `AwaitableResult` into a `collections.abc.Coroutine`.
Constant	`_F`	Undocumented
Constant	`_F1`	Undocumented
Constant	`_F2`	Undocumented
Constant	`_S`	Undocumented
Constant	`_S1`	Undocumented
Constant	`_S2`	Undocumented

def construct_failure(value: _F) -> AwaitableFailure[_F]: (source) ¶

Create an AwaitableFailure object from a value.

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from trcks.fp.monads import awaitable_result as ar
>>> a_rslt = ar.construct_failure("not found")
>>> isinstance(a_rslt, Awaitable)
True
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('failure', 'not found')

Parameters
value:`_F`	Value to be wrapped in an `AwaitableFailure` object.

Returns
`AwaitableFailure[_F]`	A new `AwaitableFailure` instance containing the given value.

def construct_failure_from_awaitable(awtbl: Awaitable[_F]) -> AwaitableFailure[_F]: (source) ¶

Create an AwaitableFailure object from an Awaitable object.

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from http import HTTPStatus
>>> from trcks.fp.monads import awaitable_result as ar
>>> async def get_status() -> HTTPStatus:
...     await asyncio.sleep(0.001)
...     return HTTPStatus.NOT_FOUND
...
>>> awaitable_status = get_status()
>>> isinstance(awaitable_status, Awaitable)
True
>>> a_rslt = ar.construct_failure_from_awaitable(awaitable_status)
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('failure', <HTTPStatus.NOT_FOUND: 404>)

Parameters
awtbl:`Awaitable[_F]`	`Awaitable` object to be wrapped in an `AwaitableFailure` object.

Returns
`AwaitableFailure[_F]`	A new `AwaitableFailure` instance containing the value of the given `Awaitable` object.

def construct_from_result(rslt: Result[_F, _S]) -> AwaitableResult[_F, _S]: (source) ¶

Create an AwaitableResult object from a Result object.

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from trcks.fp.monads import awaitable_result as ar
>>> a_rslt = ar.construct_from_result(("failure", "not found"))
>>> isinstance(a_rslt, Awaitable)
True
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('failure', 'not found')

Parameters
rslt:`Result[_F, _S]`	`Result` object to be wrapped in an `AwaitableResult` object.

Returns
`AwaitableResult[_F, _S]`	A new `AwaitableResult` instance containing the value of the given `Result` object.

def construct_success(value: _S) -> AwaitableSuccess[_S]: (source) ¶

Create an AwaitableSuccess object from a value.

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from trcks.fp.monads import awaitable_result as ar
>>> a_rslt = ar.construct_success(42)
>>> isinstance(a_rslt, Awaitable)
True
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('success', 42)

Parameters
value:`_S`	Value to be wrapped in an `AwaitableSuccess` object.

Returns
`AwaitableSuccess[_S]`	A new `AwaitableSuccess` instance containing the given value.

def construct_success_from_awaitable(awtbl: Awaitable[_S]) -> AwaitableSuccess[_S]: (source) ¶

Create an AwaitableSuccess object from an Awaitable object.

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from trcks.fp.monads import awaitable_result as ar
>>> async def read_from_disk() -> int:
...     await asyncio.sleep(0.001)
...     return "Hello, world!"
...
>>> awaitable_str = read_from_disk()
>>> isinstance(awaitable_str, Awaitable)
True
>>> a_rslt = ar.construct_success_from_awaitable(awaitable_str)
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('success', 'Hello, world!')

Parameters
awtbl:`Awaitable[_S]`	`Awaitable` object to be wrapped in an `AwaitableSuccess` object.

Returns
`AwaitableSuccess[_S]`	A new `AwaitableSuccess` instance containing the value of the given `Awaitable` object.

def map_failure(f: Callable[[_F1], _F2]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1]]: (source) ¶

Create function that maps AwaitableFailure to AwaitableFailure values.

AwaitableSuccess values are left unchanged.

Example

>>> import asyncio
>>> from trcks import AwaitableResult
>>> from trcks.fp.monads import awaitable_result as ar
>>> add_prefix_to_failure = ar.map_failure(lambda s: f"Prefix: {s}")
>>> a_rslt_1: AwaitableResult[str, float] = add_prefix_to_failure(
...     ar.construct_failure("negative value")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'Prefix: negative value')
>>> a_rslt_2: AwaitableResult[str, float] = add_prefix_to_failure(
...     ar.construct_success(25.0)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 25.0)

Parameters
f:`Callable[[_F1], _F2]`	Function to apply to the `AwaitableFailure` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1]]`	Maps `AwaitableFailure` values to `AwaitableFailure` values according to the given function and leaves `AwaitableSuccess` values unchanged.

def map_failure_to_awaitable(f: Callable[[_F1], Awaitable[_F2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1]]: (source) ¶

Create function that maps AwaitableFailure to AwaitableFailure values.

AwaitableSuccess values are left unchanged.

Example

>>> import asyncio
>>> from trcks import AwaitableResult
>>> from trcks.fp.monads import awaitable_result as ar
>>> async def slowly_add_prefix(s: str) -> str:
...     await asyncio.sleep(0.001)
...     return f"Prefix: {s}"
...
>>> slowly_add_prefix_to_failure = ar.map_failure_to_awaitable(
...     slowly_add_prefix
... )
>>> a_rslt_1: AwaitableResult[str, float] = slowly_add_prefix_to_failure(
...     ar.construct_failure("negative value")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'Prefix: negative value')
>>> a_rslt_2: AwaitableResult[str, float] = slowly_add_prefix_to_failure(
...     ar.construct_success(25.0)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 25.0)

Parameters
f:`Callable[[_F1], Awaitable[_F2]]`	Asynchronous function to apply to the `AwaitableFailure` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1]]`	Maps `AwaitableFailure` values to `AwaitableFailure` values according to the given asynchronous function and leaves `AwaitableSuccess` values unchanged.

def map_failure_to_awaitable_result(f: Callable[[_F1], AwaitableResult[_F2, _S2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1 | _S2]]: (source) ¶

Create function that maps AwaitableFailure values to AwaitableResult values.

AwaitableSuccess values are left unchanged.

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> 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
...
>>> slowly_replace_not_found = ar.map_failure_to_awaitable_result(
...     _slowly_replace_not_found
... )
>>>
>>> a_rslt_1 = slowly_replace_not_found(ar.construct_failure("not found"))
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('success', 0.0)
>>> a_rslt_2 = slowly_replace_not_found(ar.construct_failure("other failure"))
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('failure', 'other failure')
>>> a_rslt_3 = slowly_replace_not_found(ar.construct_success(25.0))
>>> asyncio.run(ar.to_coroutine_result(a_rslt_3))
('success', 25.0)

Parameters
f:`Callable[[_F1], AwaitableResult[_F2, _S2]]`	Asynchronous function to apply to the `AwaitableFailure` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1 \| _S2]]`	Maps `AwaitableFailure` values to `AwaitableFailure` and `AwaitableSuccess` values according to the given asynchronous function and leaves `AwaitableSuccess` values unchanged.

def map_failure_to_result(f: Callable[[_F1], Result[_F2, _S2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1 | _S2]]: (source) ¶

Create function that maps AwaitableFailure values to AwaitableResult values.

AwaitableSuccess values are left unchanged.

Example

>>> import asyncio
>>> from trcks import AwaitableResult
>>> from trcks.fp.monads import awaitable_result as ar
>>> replace_not_found_by_default_value = ar.map_failure_to_result(
...     lambda s: ("success", 0.0) if s == "not found" else ("failure", s)
... )
>>> a_rslt_1: AwaitableResult[str, float] = replace_not_found_by_default_value(
...     ar.construct_failure("not found")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('success', 0.0)
>>> a_rslt_2: AwaitableResult[str, float] = replace_not_found_by_default_value(
...     ar.construct_failure("other failure")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('failure', 'other failure')
>>> a_rslt_3: AwaitableResult[str, float] = replace_not_found_by_default_value(
...     ar.construct_success(25.0)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_3))
('success', 25.0)

Parameters
f:`Callable[[_F1], Result[_F2, _S2]]`	Function to apply to the `AwaitableFailure` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F2, _S1 \| _S2]]`	Maps `AwaitableFailure` values to `AwaitableResult` values according to the given function and leaves `AwaitableSuccess` values unchanged.

def map_success(f: Callable[[_S1], _S2]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1, _S2]]: (source) ¶

Creates function that maps AwaitableSuccess to AwaitableSuccess values.

AwaitableFailure values are left unchanged.

Example

>>> import asyncio
>>> from trcks import AwaitableResult
>>> from trcks.fp.monads import awaitable_result as ar
>>> def increase(n: int) -> int:
...     return n + 1
...
>>> increase_success = ar.map_success(increase)
>>> a_rslt_1: AwaitableResult[str, int] = increase_success(
...     ar.construct_failure("not found")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'not found')
>>> a_rslt_2: AwaitableResult[str, int] = increase_success(
...     ar.construct_success(42)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 43)

Parameters
f:`Callable[[_S1], _S2]`	Function to apply to the `AwaitableSuccess` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1, _S2]]`	Leaves `AwaitableFailure` values unchanged and maps `AwaitableSuccess` values to new `AwaitableSuccess` values according to the given function.

def map_success_to_awaitable(f: Callable[[_S1], Awaitable[_S2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1, _S2]]: (source) ¶

Creates function that maps AwaitableSuccess to AwaitableSuccess values.

AwaitableFailure values are left unchanged.

Example

>>> import asyncio
>>>
>>> from trcks import AwaitableResult
>>> from trcks.fp.monads import awaitable_result as ar
>>>
>>>
>>> async def increment_slowly(n: int) -> int:
...     return n + 1
...
>>> increase_success = ar.map_success_to_awaitable(increment_slowly)
>>>
>>> a_rslt_1: AwaitableResult[str, int] = increase_success(
...     ar.construct_failure("not found")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'not found')
>>>
>>> a_rslt_2: AwaitableResult[str, int] = increase_success(
...     ar.construct_success(42)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 43)

Parameters
f:`Callable[[_S1], Awaitable[_S2]]`	Asynchronous function to apply to the `AwaitableSuccess` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1, _S2]]`	Leaves `AwaitableFailure` values unchanged and maps `AwaitableSuccess` values to new `AwaitableSuccess` values according to the given function.

def map_success_to_awaitable_result(f: Callable[[_S1], AwaitableResult[_F2, _S2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1 | _F2, _S2]]: (source) ¶

Create function that maps AwaitableSuccess values to AwaitableResult values.

AwaitableFailure values are left unchanged.

Example

>>> import asyncio
>>> import math
>>> from trcks import AwaitableResult, Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> 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)
...
>>> get_square_root_slowly = ar.map_success_to_awaitable_result(
...     _get_square_root_slowly
... )
>>> a_rslt_1: AwaitableResult[str, float] = get_square_root_slowly(
...     ar.construct_failure("not found")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'not found')
>>> a_rslt_2: AwaitableResult[str, float] = get_square_root_slowly(
...     ar.construct_success(25.0)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 5.0)

Parameters
f:`Callable[[_S1], AwaitableResult[_F2, _S2]]`	Asynchronous function to apply to the `AwaitableSuccess` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1 \| _F2, _S2]]`	Leaves `AwaitableFailure` values unchanged and maps `AwaitableSuccess` values to `AwaitableFailure` and `AwaitableSuccess` values according to the given asynchronous function.

def map_success_to_result(f: Callable[[_S1], Result[_F2, _S2]]) -> Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1 | _F2, _S2]]: (source) ¶

Creates function that maps AwaitableSuccess values to AwaitableResult values.

AwaitableFailure values are left unchanged.

Example

>>> import asyncio
>>> import math
>>> from trcks import Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> def _get_square_root_slowly(x: float) -> Result[str, float]:
...     if x < 0:
...         return "failure", "negative value"
...     return "success", math.sqrt(x)
...
>>> get_square_root_slowly = ar.map_success_to_result(
...     _get_square_root_slowly
... )
>>> a_rslt_1 = get_square_root_slowly(
...     ar.construct_failure("not found")
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'not found')
>>> a_rslt_2 = get_square_root_slowly(
...     ar.construct_success(25.0)
... )
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
('success', 5.0)

Parameters
f:`Callable[[_S1], Result[_F2, _S2]]`	Function to apply to the `AwaitableSuccess` values.

Returns
`Callable[[AwaitableResult[_F1, _S1]], AwaitableResult[_F1 \| _F2, _S2]]`	Leaves `AwaitableFailure` values unchanged and maps `AwaitableSuccess` values to `AwaitableFailure` and `AwaitableSuccess` values according to the given function.

async def to_coroutine_result(a_rslt: AwaitableResult[_F, _S]) -> Result[_F, _S]: (source) ¶

Turn an AwaitableResult into a collections.abc.Coroutine.

This is useful for functions that expect a coroutine (e.g. asyncio.run).

Example

>>> import asyncio
>>> from trcks import Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> asyncio.set_event_loop(asyncio.new_event_loop())
>>> future = asyncio.Future[Result[str, int]]()
>>> future.set_result(("success", 42))
>>> future
<Future finished result=('success', 42)>
>>> coro = ar.to_coroutine_result(future)
>>> coro
<coroutine object to_coroutine_result at ...>
>>> asyncio.run(coro)
('success', 42)

Parameters
a_rslt:`AwaitableResult[_F, _S]`	The `AwaitableResult` to be transformed into a `collections.abc.Coroutine`.

Returns
`Result[_F, _S]`	The given `AwaitableResult` transformed into a `collections.abc.Coroutine`.

_F = (source) ¶

Undocumented

Value

TypeVar('_F')

_F1 = (source) ¶

Undocumented

Value

TypeVar('_F1')

_F2 = (source) ¶

Undocumented

Value

TypeVar('_F2')

_S = (source) ¶

Undocumented

Value

TypeVar('_S')

_S1 = (source) ¶

Undocumented

Value

TypeVar('_S1')

_S2 = (source) ¶

Undocumented

Value

TypeVar('_S2')

trcks.fp.monads.awaitable_result

`trcks.fp.monads.awaitable_result`