Skip to content

trcks.fp.monads.identity

Functions for the identity monad.

Provides utilities for functional composition of synchronous functions.

map_to_awaitable(callable_, /, *args, **kwargs)

Create function that maps a plain value to an collections.abc.Awaitable value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Awaitable[_T2]]) –

    Asynchronous function to apply to the given value.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> import asyncio
>>> from trcks.fp.monads import awaitable as a
>>> from trcks.fp.monads import identity as i
>>> async def stringify_slowly(o: object) -> str:
...     await asyncio.sleep(0.001)
...     return str(o)
...
>>> stringify = i.map_to_awaitable(stringify_slowly)
>>> awtbl = stringify(3.14)
>>> asyncio.run(a.to_coroutine(awtbl))
'3.14'
Source code in src/trcks/fp/monads/identity.py
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def map_to_awaitable(
    callable_: Callable[Concatenate[_T1, _P], Awaitable[_T2]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], Awaitable[_T2]]:
    """Create function that maps a plain value
    to an [collections.abc.Awaitable][] value.

    Args:
        callable_: Asynchronous function to apply to the given value.
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [collections.abc.Awaitable][] values
            according to the given asynchronous function.

    Examples:
        >>> import asyncio
        >>> from trcks.fp.monads import awaitable as a
        >>> from trcks.fp.monads import identity as i
        >>> async def stringify_slowly(o: object) -> str:
        ...     await asyncio.sleep(0.001)
        ...     return str(o)
        ...
        >>> stringify = i.map_to_awaitable(stringify_slowly)
        >>> awtbl = stringify(3.14)
        >>> asyncio.run(a.to_coroutine(awtbl))
        '3.14'
    """
    return compose(a.construct, a.map_to_awaitable(callable_, *args, **kwargs))

map_to_awaitable_iterable(callable_, /, *args, **kwargs)

Create function that maps a plain value to a trcks.AwaitableIterable and flattens the result.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], AwaitableIterable[_T2]]) –

    Asynchronous function to apply to the given value, returning a trcks.AwaitableIterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> import asyncio
>>> from trcks.fp.monads import awaitable_tuple as at
>>> from trcks.fp.monads import identity as i
>>> async def slowly_duplicate(n: int) -> tuple[int, int]:
...     await asyncio.sleep(0.001)
...     return n, n
...
>>> duplicate = i.map_to_awaitable_iterable(slowly_duplicate)
>>> a_tpl = duplicate(7)
>>> asyncio.run(at.to_coroutine_tuple(a_tpl))
(7, 7)
Source code in src/trcks/fp/monads/identity.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def map_to_awaitable_iterable(
    callable_: Callable[Concatenate[_T1, _P], AwaitableIterable[_T2]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableTuple[_T2]]:
    """Create function that maps a plain value
    to a [trcks.AwaitableIterable][] and flattens the result.

    Args:
        callable_: Asynchronous function to apply to the given value,
            returning a [trcks.AwaitableIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [trcks.AwaitableTuple][]s of varying length
            according to the given asynchronous function.

    Examples:
        >>> import asyncio
        >>> from trcks.fp.monads import awaitable_tuple as at
        >>> from trcks.fp.monads import identity as i
        >>> async def slowly_duplicate(n: int) -> tuple[int, int]:
        ...     await asyncio.sleep(0.001)
        ...     return n, n
        ...
        >>> duplicate = i.map_to_awaitable_iterable(slowly_duplicate)
        >>> a_tpl = duplicate(7)
        >>> asyncio.run(at.to_coroutine_tuple(a_tpl))
        (7, 7)
    """
    return compose(
        at.construct, at.map_to_awaitable_iterable(callable_, *args, **kwargs)
    )

map_to_awaitable_result(callable_, /, *args, **kwargs)

Create function that maps a plain value to a trcks.AwaitableResult value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], AwaitableResult[_F, _S]]) –

    Asynchronous function to apply to the given value, returning a trcks.AwaitableResult.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> import asyncio
>>> from trcks import Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> from trcks.fp.monads import identity as i
>>> 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
...
>>> assert_non_negative = i.map_to_awaitable_result(
...     slowly_assert_non_negative
... )
>>> a_rslt = assert_non_negative(42.0)
>>> asyncio.run(ar.to_coroutine_result(a_rslt))
('success', 42.0)
Source code in src/trcks/fp/monads/identity.py
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def map_to_awaitable_result(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResult[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResult[_F, _S]]:
    """Create function that maps a plain value to a [trcks.AwaitableResult][] value.

    Args:
        callable_: Asynchronous function to apply to the given value,
            returning a [trcks.AwaitableResult][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [trcks.AwaitableResult][] values
            according to the given asynchronous function.

    Examples:
        >>> import asyncio
        >>> from trcks import Result
        >>> from trcks.fp.monads import awaitable_result as ar
        >>> from trcks.fp.monads import identity as i
        >>> 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
        ...
        >>> assert_non_negative = i.map_to_awaitable_result(
        ...     slowly_assert_non_negative
        ... )
        >>> a_rslt = assert_non_negative(42.0)
        >>> asyncio.run(ar.to_coroutine_result(a_rslt))
        ('success', 42.0)
    """
    c: Composable2[
        [_T1],
        AwaitableResult[Never, _T1],
        AwaitableResult[_F, _S],
    ] = (
        ar.construct_success,
        ar.map_success_to_awaitable_result(callable_, *args, **kwargs),
    )
    return compose(*c)

map_to_awaitable_result_iterable(callable_, /, *args, **kwargs)

Create function that maps a plain value to a trcks.AwaitableResultIterable and flattens the result.

Parameters:

Returns:

Examples:

>>> import asyncio
>>> from trcks import AwaitableResultTuple
>>> from trcks.fp.monads import awaitable_result_tuple as art
>>> from trcks.fp.monads import identity as i
>>> async def validate(x: float) -> AwaitableResultTuple[str, float]:
...     await asyncio.sleep(0.001)
...     if x < 0:
...         return "failure", "negative value"
...     return "success", (x, x * 2)
...
>>> validate_value = i.map_to_awaitable_result_iterable(validate)
>>> a_r_tpl = validate_value(5.0)
>>> asyncio.run(art.to_coroutine_result_tuple(a_r_tpl))
('success', (5.0, 10.0))
Source code in src/trcks/fp/monads/identity.py
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def map_to_awaitable_result_iterable(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResultIterable[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResultTuple[_F, _S]]:
    """Create function that maps a plain value
    to a [trcks.AwaitableResultIterable][] and flattens the result.

    Args:
        callable_: Asynchronous function to apply to the given value,
            returning a [trcks.AwaitableResultIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [trcks.AwaitableResultTuple][]s with

            - the [trcks.Failure][] returned by the function, or
            - a [trcks.SuccessTuple][] containing the elements of
                the [trcks.SuccessIterable][] returned by the function.

    Examples:
        >>> import asyncio
        >>> from trcks import AwaitableResultTuple
        >>> from trcks.fp.monads import awaitable_result_tuple as art
        >>> from trcks.fp.monads import identity as i
        >>> async def validate(x: float) -> AwaitableResultTuple[str, float]:
        ...     await asyncio.sleep(0.001)
        ...     if x < 0:
        ...         return "failure", "negative value"
        ...     return "success", (x, x * 2)
        ...
        >>> validate_value = i.map_to_awaitable_result_iterable(validate)
        >>> a_r_tpl = validate_value(5.0)
        >>> asyncio.run(art.to_coroutine_result_tuple(a_r_tpl))
        ('success', (5.0, 10.0))
    """
    c: Composable2[
        [_T1],
        AwaitableResultTuple[Never, _T1],
        AwaitableResultTuple[_F, _S],
    ] = (
        art.construct_successes,
        art.map_successes_to_awaitable_result_iterable(callable_, *args, **kwargs),
    )
    return compose(*c)

map_to_awaitable_result_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.map_to_awaitable_result_iterable.

Source code in src/trcks/fp/monads/identity.py
246
247
248
249
250
251
252
253
254
255
256
257
258
@deprecated("Use map_to_awaitable_result_iterable instead")
def map_to_awaitable_result_tuple(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResultTuple[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResultTuple[_F, _S]]:
    """Deprecated alias for
    [trcks.fp.monads.identity.map_to_awaitable_result_iterable][].
    """
    return map_to_awaitable_result_iterable(
        callable_, *args, **kwargs
    )  # pragma: no cover

map_to_awaitable_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.map_to_awaitable_iterable.

Source code in src/trcks/fp/monads/identity.py
261
262
263
264
265
266
267
268
269
@deprecated("Use map_to_awaitable_iterable instead")
def map_to_awaitable_tuple(
    callable_: Callable[Concatenate[_T1, _P], AwaitableTuple[_T2]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableTuple[_T2]]:
    """Deprecated alias for [trcks.fp.monads.identity.map_to_awaitable_iterable][]."""
    return map_to_awaitable_iterable(callable_, *args, **kwargs)  # pragma: no cover

map_to_iterable(callable_, /, *args, **kwargs)

Create function that maps a plain value to a collections.abc.Iterable and flattens the result.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Iterable[_T2]]) –

    Function to apply to the given value, returning a collections.abc.Iterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T1], tuple[_T2, ...]] –

    Maps plain values to homogeneous tuples of varying length according to the given function.

Examples:

>>> from trcks.fp.monads import identity as i
>>> def duplicate(n: int) -> tuple[int, int]:
...     return n, n
...
>>> duplicate_value = i.map_to_iterable(duplicate)
>>> duplicate_value(3)
(3, 3)
Source code in src/trcks/fp/monads/identity.py
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def map_to_iterable(
    callable_: Callable[Concatenate[_T1, _P], Iterable[_T2]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], tuple[_T2, ...]]:
    """Create function that maps a plain value
    to a [collections.abc.Iterable][] and flattens the result.

    Args:
        callable_: Function to apply to the given value,
            returning a [collections.abc.Iterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to homogeneous [tuple][]s of varying length
            according to the given function.

    Examples:
        >>> from trcks.fp.monads import identity as i
        >>> def duplicate(n: int) -> tuple[int, int]:
        ...     return n, n
        ...
        >>> duplicate_value = i.map_to_iterable(duplicate)
        >>> duplicate_value(3)
        (3, 3)
    """
    return compose(t.construct, t.map_to_iterable(callable_, *args, **kwargs))

map_to_result(callable_, /, *args, **kwargs)

Create function that maps a plain value to a trcks.Result value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Result[_F, _S]]) –

    Function to apply to the given value, returning a trcks.Result.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> from trcks.fp.monads import identity as i
>>> assert_non_negative = i.map_to_result(
...     lambda n: ("success", n)
...     if n >= 0
...     else ("failure", "negative value")
... )
>>> assert_non_negative(-1)
('failure', 'negative value')
Source code in src/trcks/fp/monads/identity.py
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def map_to_result(
    callable_: Callable[Concatenate[_T1, _P], Result[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], Result[_F, _S]]:
    """Create function that maps a plain value to a [trcks.Result][] value.

    Args:
        callable_: Function to apply to the given value, returning a [trcks.Result][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [trcks.Result][] values
            according to the given function.

    Examples:
        >>> from trcks.fp.monads import identity as i
        >>> assert_non_negative = i.map_to_result(
        ...     lambda n: ("success", n)
        ...     if n >= 0
        ...     else ("failure", "negative value")
        ... )
        >>> assert_non_negative(-1)
        ('failure', 'negative value')
    """
    c: Composable2[
        [_T1],
        Result[Never, _T1],
        Result[_F, _S],
    ] = (r.construct_success, r.map_success_to_result(callable_, *args, **kwargs))
    return compose(*c)

map_to_result_iterable(callable_, /, *args, **kwargs)

Create function that maps a plain value to a trcks.ResultIterable and flattens the result.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], ResultIterable[_F, _S]]) –

    Function to apply to the given value, returning a trcks.ResultIterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> from trcks.fp.monads import identity as i
>>> duplicate_value = i.map_to_result_iterable(
...     lambda n: ("success", (n, n))
...     if n >= 0
...     else ("failure", "negative value")
... )
>>> duplicate_value(-1)
('failure', 'negative value')
Source code in src/trcks/fp/monads/identity.py
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def map_to_result_iterable(
    callable_: Callable[Concatenate[_T1, _P], ResultIterable[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], ResultTuple[_F, _S]]:
    """Create function that maps a plain value
    to a [trcks.ResultIterable][] and flattens the result.

    Args:
        callable_: Function to apply to the given value,
            returning a [trcks.ResultIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Maps plain values to [trcks.ResultTuple][]s with

            - the [trcks.Failure][] returned by the function, or
            - a [trcks.SuccessTuple][] containing the elements of
                the [trcks.SuccessIterable][] returned by the function.

    Examples:
        >>> from trcks.fp.monads import identity as i
        >>> duplicate_value = i.map_to_result_iterable(
        ...     lambda n: ("success", (n, n))
        ...     if n >= 0
        ...     else ("failure", "negative value")
        ... )
        >>> duplicate_value(-1)
        ('failure', 'negative value')
    """
    c: Composable2[
        [_T1],
        ResultTuple[Never, _T1],
        ResultTuple[_F, _S],
    ] = (
        rt.construct_successes,
        rt.map_successes_to_result_iterable(callable_, *args, **kwargs),
    )
    return compose(*c)

map_to_result_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.map_to_result_iterable.

Source code in src/trcks/fp/monads/identity.py
387
388
389
390
391
392
393
394
395
@deprecated("Use map_to_result_iterable instead")
def map_to_result_tuple(
    callable_: Callable[Concatenate[_T1, _P], ResultTuple[_F, _S]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], ResultTuple[_F, _S]]:
    """Deprecated alias for [trcks.fp.monads.identity.map_to_result_iterable][]."""
    return map_to_result_iterable(callable_, *args, **kwargs)  # pragma: no cover

map_to_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.map_to_iterable.

Source code in src/trcks/fp/monads/identity.py
398
399
400
401
402
403
404
405
406
@deprecated("Use map_to_iterable instead")
def map_to_tuple(
    callable_: Callable[Concatenate[_T1, _P], tuple[_T2, ...]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], tuple[_T2, ...]]:
    """Deprecated alias for [trcks.fp.monads.identity.map_to_iterable][]."""
    return map_to_iterable(callable_, *args, **kwargs)  # pragma: no cover

tap(callable_, /, *args, **kwargs)

Turn synchronous function into a function that returns its input.

Parameters:

  • callable_ (Callable[Concatenate[_T, _P], object]) –

    The synchronous function to be transformed into a function that returns its input.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T], _T] –

    The given function transformed into a function that returns its input.

Examples:

>>> from collections.abc import Callable
>>> from trcks.fp.monads import identity as i
>>> log_and_pass_on: Callable[[object], object] = i.tap(
...     lambda o: print(f"Received object {o}.")
... )
>>> output = log_and_pass_on(42)
Received object 42.
>>> output
42
Source code in src/trcks/fp/_monads/identity.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def tap(
    callable_: Callable[Concatenate[_T, _P], object],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T], _T]:
    """Turn synchronous function into a function that returns its input.

    Args:
        callable_:
            The synchronous function to be transformed into
            a function that returns its input.
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        The given function transformed into a function that returns its input.

    Examples:
        >>> from collections.abc import Callable
        >>> from trcks.fp.monads import identity as i
        >>> log_and_pass_on: Callable[[object], object] = i.tap(
        ...     lambda o: print(f"Received object {o}.")
        ... )
        >>> output = log_and_pass_on(42)
        Received object 42.
        >>> output
        42
    """

    def bypassed_callable(value: _T) -> _T:
        _ = callable_(value, *args, **kwargs)
        return value

    return bypassed_callable

tap_to_awaitable(callable_, /, *args, **kwargs)

Create function that applies an asynchronous side effect to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Awaitable[object]]) –

    Asynchronous side effect to apply to the given value.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> import asyncio
>>> from trcks.fp.monads import awaitable as a
>>> from trcks.fp.monads import identity as i
>>> async def write_to_disk(s: str) -> None:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' to disk.")
...
>>> write_to_disk_tapped = i.tap_to_awaitable(write_to_disk)
>>> awtbl = write_to_disk_tapped("Hello, world!")
>>> value = asyncio.run(a.to_coroutine(awtbl))
Wrote 'Hello, world!' to disk.
>>> value
'Hello, world!'
Source code in src/trcks/fp/monads/identity.py
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
def tap_to_awaitable(
    callable_: Callable[Concatenate[_T1, _P], Awaitable[object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], Awaitable[_T1]]:
    """Create function that applies an asynchronous side effect to a plain value.

    Args:
        callable_: Asynchronous side effect to apply to the given value.
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns an [collections.abc.Awaitable][] of the original value.

    Examples:
        >>> import asyncio
        >>> from trcks.fp.monads import awaitable as a
        >>> from trcks.fp.monads import identity as i
        >>> async def write_to_disk(s: str) -> None:
        ...     await asyncio.sleep(0.001)
        ...     print(f"Wrote '{s}' to disk.")
        ...
        >>> write_to_disk_tapped = i.tap_to_awaitable(write_to_disk)
        >>> awtbl = write_to_disk_tapped("Hello, world!")
        >>> value = asyncio.run(a.to_coroutine(awtbl))
        Wrote 'Hello, world!' to disk.
        >>> value
        'Hello, world!'
    """
    return compose(a.construct, a.tap_to_awaitable(callable_, *args, **kwargs))

tap_to_awaitable_iterable(callable_, /, *args, **kwargs)

Create function that applies an asynchronous side effect with return type trcks.AwaitableIterable to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], AwaitableIterable[object]]) –

    Asynchronous side effect to apply to the given value, returning a trcks.AwaitableIterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

Examples:

>>> import asyncio
>>> from trcks.fp.monads import awaitable_tuple as at
>>> from trcks.fp.monads import identity as i
>>> async def write_to_disk(n: int) -> tuple[str, str]:
...     await asyncio.sleep(0.001)
...     print(f"Wrote {n} to disk.")
...     return "left", "right"
...
>>> write_to_disk_tapped = i.tap_to_awaitable_iterable(write_to_disk)
>>> a_tpl = write_to_disk_tapped(3)
>>> asyncio.run(at.to_coroutine_tuple(a_tpl))
Wrote 3 to disk.
(3, 3)
Source code in src/trcks/fp/monads/identity.py
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def tap_to_awaitable_iterable(
    callable_: Callable[Concatenate[_T1, _P], AwaitableIterable[object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableTuple[_T1]]:
    """Create function that applies an asynchronous side effect
    with return type [trcks.AwaitableIterable][] to a plain value.

    Args:
        callable_: Asynchronous side effect to apply to the given value,
            returning a [trcks.AwaitableIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a [trcks.AwaitableTuple][] of the original value,
            repeated once per element in the side effect output.

    Examples:
        >>> import asyncio
        >>> from trcks.fp.monads import awaitable_tuple as at
        >>> from trcks.fp.monads import identity as i
        >>> async def write_to_disk(n: int) -> tuple[str, str]:
        ...     await asyncio.sleep(0.001)
        ...     print(f"Wrote {n} to disk.")
        ...     return "left", "right"
        ...
        >>> write_to_disk_tapped = i.tap_to_awaitable_iterable(write_to_disk)
        >>> a_tpl = write_to_disk_tapped(3)
        >>> asyncio.run(at.to_coroutine_tuple(a_tpl))
        Wrote 3 to disk.
        (3, 3)
    """
    return compose(
        at.construct, at.tap_to_awaitable_iterable(callable_, *args, **kwargs)
    )

tap_to_awaitable_result(callable_, /, *args, **kwargs)

Create function that applies an asynchronous side effect with return type trcks.Result to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], AwaitableResult[_F, object]]) –

    Asynchronous side effect to apply to the given value, returning a trcks.Result.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T1], AwaitableResult[_F, _T1]] –

    Applies the given side effect to the given value and returns a trcks.AwaitableResult with

    - *the returned* [trcks.Failure][]
        if the applied side effect returns a [trcks.Failure][] or
    - *the original* value
        if the applied side effect returns a [trcks.Success][].
    

Examples:

>>> import asyncio
>>> from trcks import Result
>>> from trcks.fp.monads import awaitable_result as ar
>>> from trcks.fp.monads import identity as i
>>> async def write_to_disk(s: str, path: str) -> Result[str, None]:
...     if path != "output.txt":
...         return "failure", "write error"
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' to file {path}.")
...     return "success", None
...
>>> write_to_disk_tapped = i.tap_to_awaitable_result(
...     lambda s: write_to_disk(s, "destination.txt")
... )
>>> a_rslt_1 = write_to_disk_tapped("Hello, world!")
>>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
('failure', 'write error')
>>> write_to_disk_tapped = i.tap_to_awaitable_result(
...     lambda s: write_to_disk(s, "output.txt")
... )
>>> a_rslt_2 = write_to_disk_tapped("Hello, world!")
>>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
Wrote 'Hello, world!' to file output.txt.
('success', 'Hello, world!')
Source code in src/trcks/fp/monads/identity.py
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
def tap_to_awaitable_result(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResult[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResult[_F, _T1]]:
    """Create function that applies an asynchronous side effect
    with return type [trcks.Result][] to a plain value.

    Args:
        callable_: Asynchronous side effect to apply to the given value,
            returning a [trcks.Result][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a [trcks.AwaitableResult][] with

                - *the returned* [trcks.Failure][]
                    if the applied side effect returns a [trcks.Failure][] or
                - *the original* value
                    if the applied side effect returns a [trcks.Success][].

    Examples:
        >>> import asyncio
        >>> from trcks import Result
        >>> from trcks.fp.monads import awaitable_result as ar
        >>> from trcks.fp.monads import identity as i
        >>> async def write_to_disk(s: str, path: str) -> Result[str, None]:
        ...     if path != "output.txt":
        ...         return "failure", "write error"
        ...     await asyncio.sleep(0.001)
        ...     print(f"Wrote '{s}' to file {path}.")
        ...     return "success", None
        ...
        >>> write_to_disk_tapped = i.tap_to_awaitable_result(
        ...     lambda s: write_to_disk(s, "destination.txt")
        ... )
        >>> a_rslt_1 = write_to_disk_tapped("Hello, world!")
        >>> asyncio.run(ar.to_coroutine_result(a_rslt_1))
        ('failure', 'write error')
        >>> write_to_disk_tapped = i.tap_to_awaitable_result(
        ...     lambda s: write_to_disk(s, "output.txt")
        ... )
        >>> a_rslt_2 = write_to_disk_tapped("Hello, world!")
        >>> asyncio.run(ar.to_coroutine_result(a_rslt_2))
        Wrote 'Hello, world!' to file output.txt.
        ('success', 'Hello, world!')
    """
    c: Composable2[
        [_T1],
        AwaitableResult[Never, _T1],
        AwaitableResult[_F, _T1],
    ] = (
        ar.construct_success,
        ar.tap_success_to_awaitable_result(callable_, *args, **kwargs),
    )
    return compose(*c)

tap_to_awaitable_result_iterable(callable_, /, *args, **kwargs)

Create function that applies an asynchronous side effect with return type trcks.AwaitableResultIterable to a plain value.

Parameters:

Returns:

  • Callable[[_T1], AwaitableResultTuple[_F, _T1]] –

    Applies the given side effect to the given value and returns a trcks.AwaitableResultTuple with

    - *the returned* [trcks.Failure][]
        if the applied side effect returns a [trcks.Failure][] or
    - *the original* value repeated once per element
        in the side effect output
        if the applied side effect returns a [trcks.SuccessIterable][].
    

Examples:

>>> import asyncio
>>> from trcks import AwaitableResultTuple
>>> from trcks.fp.monads import awaitable_result_tuple as art
>>> from trcks.fp.monads import identity as i
>>> async def write_twice(s: str) -> AwaitableResultTuple[str, None]:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' twice.")
...     return "success", (None, None)
...
>>> write_twice_tapped = i.tap_to_awaitable_result_iterable(write_twice)
>>> a_r_tpl = write_twice_tapped("Hello, world!")
>>> asyncio.run(art.to_coroutine_result_tuple(a_r_tpl))
Wrote 'Hello, world!' twice.
('success', ('Hello, world!', 'Hello, world!'))
Source code in src/trcks/fp/monads/identity.py
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
def tap_to_awaitable_result_iterable(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResultIterable[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResultTuple[_F, _T1]]:
    """Create function that applies an asynchronous side effect
    with return type [trcks.AwaitableResultIterable][] to a plain value.

    Args:
        callable_: Asynchronous side effect to apply to the given value,
            returning a [trcks.AwaitableResultIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a [trcks.AwaitableResultTuple][] with

                - *the returned* [trcks.Failure][]
                    if the applied side effect returns a [trcks.Failure][] or
                - *the original* value repeated once per element
                    in the side effect output
                    if the applied side effect returns a [trcks.SuccessIterable][].

    Examples:
        >>> import asyncio
        >>> from trcks import AwaitableResultTuple
        >>> from trcks.fp.monads import awaitable_result_tuple as art
        >>> from trcks.fp.monads import identity as i
        >>> async def write_twice(s: str) -> AwaitableResultTuple[str, None]:
        ...     await asyncio.sleep(0.001)
        ...     print(f"Wrote '{s}' twice.")
        ...     return "success", (None, None)
        ...
        >>> write_twice_tapped = i.tap_to_awaitable_result_iterable(write_twice)
        >>> a_r_tpl = write_twice_tapped("Hello, world!")
        >>> asyncio.run(art.to_coroutine_result_tuple(a_r_tpl))
        Wrote 'Hello, world!' twice.
        ('success', ('Hello, world!', 'Hello, world!'))
    """
    c: Composable2[
        [_T1],
        AwaitableResultTuple[Never, _T1],
        AwaitableResultTuple[_F, _T1],
    ] = (
        art.construct_successes,
        art.tap_successes_to_awaitable_result_iterable(callable_, *args, **kwargs),
    )
    return compose(*c)

tap_to_awaitable_result_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.tap_to_awaitable_result_iterable.

Source code in src/trcks/fp/monads/identity.py
605
606
607
608
609
610
611
612
613
614
615
616
617
@deprecated("Use tap_to_awaitable_result_iterable instead")
def tap_to_awaitable_result_tuple(
    callable_: Callable[Concatenate[_T1, _P], AwaitableResultTuple[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableResultTuple[_F, _T1]]:
    """Deprecated alias for
    [trcks.fp.monads.identity.tap_to_awaitable_result_iterable][].
    """
    return tap_to_awaitable_result_iterable(
        callable_, *args, **kwargs
    )  # pragma: no cover

tap_to_awaitable_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.tap_to_awaitable_iterable.

Source code in src/trcks/fp/monads/identity.py
620
621
622
623
624
625
626
627
628
@deprecated("Use tap_to_awaitable_iterable instead")
def tap_to_awaitable_tuple(
    callable_: Callable[Concatenate[_T1, _P], AwaitableTuple[object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], AwaitableTuple[_T1]]:
    """Deprecated alias for [trcks.fp.monads.identity.tap_to_awaitable_iterable][]."""
    return tap_to_awaitable_iterable(callable_, *args, **kwargs)  # pragma: no cover

tap_to_iterable(callable_, /, *args, **kwargs)

Create function that applies a side effect returning a collections.abc.Iterable to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Iterable[object]]) –

    Side effect to apply to the given value, returning a collections.abc.Iterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T1], tuple[_T1, ...]] –

    Applies the given side effect to the given value and returns a homogeneous tuple of the original value, repeated once per element in the side effect output.

Examples:

>>> from trcks.fp.monads import identity as i
>>> def write_to_disk(n: int) -> tuple[str, str]:
...     print(f"Wrote {n} to disk.")
...     return "left", "right"
...
>>> write_to_disk_tapped = i.tap_to_iterable(write_to_disk)
>>> write_to_disk_tapped(3)
Wrote 3 to disk.
(3, 3)
Source code in src/trcks/fp/monads/identity.py
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
def tap_to_iterable(
    callable_: Callable[Concatenate[_T1, _P], Iterable[object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], tuple[_T1, ...]]:
    """Create function that applies a side effect
    returning a [collections.abc.Iterable][] to a plain value.

    Args:
        callable_: Side effect to apply to the given value,
            returning a [collections.abc.Iterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a homogeneous [tuple][] of the original value,
            repeated once per element in the side effect output.

    Examples:
        >>> from trcks.fp.monads import identity as i
        >>> def write_to_disk(n: int) -> tuple[str, str]:
        ...     print(f"Wrote {n} to disk.")
        ...     return "left", "right"
        ...
        >>> write_to_disk_tapped = i.tap_to_iterable(write_to_disk)
        >>> write_to_disk_tapped(3)
        Wrote 3 to disk.
        (3, 3)
    """
    return compose(t.construct, t.tap_to_iterable(callable_, *args, **kwargs))

tap_to_result(callable_, /, *args, **kwargs)

Create function that applies a side effect with return type trcks.Result to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], Result[_F, object]]) –

    Side effect to apply to the given value, returning a trcks.Result.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T1], Result[_F, _T1]] –

    Applies the given side effect to the given value and returns a trcks.Result with

    - *the returned* [trcks.Failure][]
        if the applied side effect returns a [trcks.Failure][] or
    - *the original* value
        if the applied side effect returns a [trcks.Success][].
    

Examples:

>>> from trcks import Result
>>> from trcks.fp.monads import identity as i
>>> def print_positive_float(x: float) -> Result[str, None]:
...     if x <= 0:
...         return "failure", "not positive"
...     return "success", print(f"Positive float: {x}")
...
>>> print_positive_float_tapped = i.tap_to_result(print_positive_float)
>>> print_positive_float_tapped(-2.3)
('failure', 'not positive')
>>> print_positive_float_tapped(3.5)
Positive float: 3.5
('success', 3.5)
Source code in src/trcks/fp/monads/identity.py
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
def tap_to_result(
    callable_: Callable[Concatenate[_T1, _P], Result[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], Result[_F, _T1]]:
    """Create function that applies a side effect
    with return type [trcks.Result][] to a plain value.

    Args:
        callable_: Side effect to apply to the given value,
            returning a [trcks.Result][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a [trcks.Result][] with

                - *the returned* [trcks.Failure][]
                    if the applied side effect returns a [trcks.Failure][] or
                - *the original* value
                    if the applied side effect returns a [trcks.Success][].

    Examples:
        >>> from trcks import Result
        >>> from trcks.fp.monads import identity as i
        >>> def print_positive_float(x: float) -> Result[str, None]:
        ...     if x <= 0:
        ...         return "failure", "not positive"
        ...     return "success", print(f"Positive float: {x}")
        ...
        >>> print_positive_float_tapped = i.tap_to_result(print_positive_float)
        >>> print_positive_float_tapped(-2.3)
        ('failure', 'not positive')
        >>> print_positive_float_tapped(3.5)
        Positive float: 3.5
        ('success', 3.5)
    """
    c: Composable2[
        [_T1],
        Result[Never, _T1],
        Result[_F, _T1],
    ] = (r.construct_success, r.tap_success_to_result(callable_, *args, **kwargs))
    return compose(*c)

tap_to_result_iterable(callable_, /, *args, **kwargs)

Create function that applies a side effect with return type trcks.ResultIterable to a plain value.

Parameters:

  • callable_ (Callable[Concatenate[_T1, _P], ResultIterable[_F, object]]) –

    Side effect to apply to the given value, returning a trcks.ResultIterable.

  • *args (args, default: () ) –

    Positional arguments to be passed to callable_.

  • **kwargs (kwargs, default: {} ) –

    Keyword arguments to be passed to callable_.

Returns:

  • Callable[[_T1], ResultTuple[_F, _T1]] –

    Applies the given side effect to the given value and returns a trcks.ResultTuple with

    - *the returned* [trcks.Failure][]
        if the applied side effect returns a [trcks.Failure][] or
    - *the original* value repeated once per element
        in the side effect output
        if the applied side effect returns a [trcks.SuccessIterable][].
    

Examples:

>>> from trcks import ResultTuple
>>> from trcks.fp.monads import identity as i
>>> def print_positive_float(x: float) -> ResultTuple[str, None]:
...     if x <= 0:
...         return "failure", "not positive"
...     return (
...         "success",
...         (print(f"Positive float: {x}"), print(f"Positive float: {x}"))
...     )
...
>>> print_positive_float_tapped = i.tap_to_result_iterable(
...     print_positive_float
... )
>>> print_positive_float_tapped(-2.3)
('failure', 'not positive')
>>> print_positive_float_tapped(3.5)
Positive float: 3.5
Positive float: 3.5
('success', (3.5, 3.5))
Source code in src/trcks/fp/monads/identity.py
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
def tap_to_result_iterable(
    callable_: Callable[Concatenate[_T1, _P], ResultIterable[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], ResultTuple[_F, _T1]]:
    """Create function that applies a side effect
    with return type [trcks.ResultIterable][] to a plain value.

    Args:
        callable_: Side effect to apply to the given value,
            returning a [trcks.ResultIterable][].
        *args:
            Positional arguments to be passed to `callable_`.
        **kwargs:
            Keyword arguments to be passed to `callable_`.

    Returns:
        Applies the given side effect to the given value and
            returns a [trcks.ResultTuple][] with

                - *the returned* [trcks.Failure][]
                    if the applied side effect returns a [trcks.Failure][] or
                - *the original* value repeated once per element
                    in the side effect output
                    if the applied side effect returns a [trcks.SuccessIterable][].

    Examples:
        >>> from trcks import ResultTuple
        >>> from trcks.fp.monads import identity as i
        >>> def print_positive_float(x: float) -> ResultTuple[str, None]:
        ...     if x <= 0:
        ...         return "failure", "not positive"
        ...     return (
        ...         "success",
        ...         (print(f"Positive float: {x}"), print(f"Positive float: {x}"))
        ...     )
        ...
        >>> print_positive_float_tapped = i.tap_to_result_iterable(
        ...     print_positive_float
        ... )
        >>> print_positive_float_tapped(-2.3)
        ('failure', 'not positive')
        >>> print_positive_float_tapped(3.5)
        Positive float: 3.5
        Positive float: 3.5
        ('success', (3.5, 3.5))
    """
    c: Composable2[
        [_T1],
        ResultTuple[Never, _T1],
        ResultTuple[_F, _T1],
    ] = (
        rt.construct_successes,
        rt.tap_successes_to_result_iterable(callable_, *args, **kwargs),
    )
    return compose(*c)

tap_to_result_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.tap_to_result_iterable.

Source code in src/trcks/fp/monads/identity.py
775
776
777
778
779
780
781
782
783
@deprecated("Use tap_to_result_iterable instead")
def tap_to_result_tuple(
    callable_: Callable[Concatenate[_T1, _P], ResultTuple[_F, object]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], ResultTuple[_F, _T1]]:
    """Deprecated alias for [trcks.fp.monads.identity.tap_to_result_iterable][]."""
    return tap_to_result_iterable(callable_, *args, **kwargs)  # pragma: no cover

tap_to_tuple(callable_, /, *args, **kwargs)

Deprecated alias for trcks.fp.monads.identity.tap_to_iterable.

Source code in src/trcks/fp/monads/identity.py
786
787
788
789
790
791
792
793
794
@deprecated("Use tap_to_iterable instead")
def tap_to_tuple(
    callable_: Callable[Concatenate[_T1, _P], tuple[object, ...]],
    /,
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> Callable[[_T1], tuple[_T1, ...]]:
    """Deprecated alias for [trcks.fp.monads.identity.tap_to_iterable][]."""
    return tap_to_iterable(callable_, *args, **kwargs)  # pragma: no cover