Skip to content

trcks.fp.monads.result

Monadic functions for trcks.Result.

Provides utilities for functional composition of synchronous trcks.Result-returning functions.

Example

Create and process a value of type trcks.Result:

>>> import math
>>> from trcks.fp.composition import pipe
>>> from trcks.fp.monads import result as r
>>> rslt = pipe(
...     (
...         r.construct_success(1_000_000.0),
...         r.tap_success(lambda x: print(f"Processing value {x} ...")),
...         r.map_success_to_result(
...             lambda x: (
...                 ("success", math.sqrt(x))
...                 if x >= 0
...                 else ("failure", "negative value")
...             )
...         ),
...         r.tap_success_to_result(
...             lambda x: (
...                 ("success", print(f"Wrote result {x} to disk."))
...                 if x < 100
...                 else ("failure", "out of disk space")
...             )
...         ),
...     )
... )
Processing value 1000000.0 ...
>>> rslt
('failure', 'out of disk space')

If your static type checker cannot infer the type of the argument passed to trcks.fp.composition.pipe, you can explicitly assign a type:

>>> import math
>>> from trcks import Result, Success
>>> from trcks.fp.composition import Pipeline3, pipe
>>> from trcks.fp.monads import result as r
>>> p: Pipeline3[
...     Success[float],
...     Result[str, float],
...     Result[str, float],
...     Result[str, float],
... ] = (
...     r.construct_success(1_000_000.0),
...     r.tap_success(lambda x: print(f"Processing value {x} ...")),
...     r.map_success_to_result(
...         lambda x: (
...             ("success", math.sqrt(x))
...             if x >= 0
...             else ("failure", "negative value")
...         )
...     ),
...     r.tap_success_to_result(
...         lambda x: (
...             ("success", print(f"Wrote result {x} to disk."))
...             if x < 100
...             else ("failure", "out of disk space")
...         )
...     ),
... )
>>> rslt = pipe(p)
Processing value 1000000.0 ...
>>> rslt
('failure', 'out of disk space')

construct_failure(value)

Create a trcks.Failure object from a value.

Parameters:

Returns:

Example 
>>> from trcks.fp.monads import result as r
>>> r.construct_failure(42)
('failure', 42)
Source code in src/trcks/fp/monads/result.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def construct_failure(value: _F) -> Failure[_F]:
    """Create a [trcks.Failure][] object from a value.

    Args:
        value: Value to be wrapped in a [trcks.Failure][] object.

    Returns:
        [trcks.Failure][] object containing the given value.

    Example:
        >>> from trcks.fp.monads import result as r
        >>> r.construct_failure(42)
        ('failure', 42)
    """
    return "failure", value

construct_success(value)

Create a trcks.Success object from a value.

Parameters:

Returns:

Example 
>>> from trcks.fp.monads import result as r
>>> r.construct_success(42)
('success', 42)
Source code in src/trcks/fp/monads/result.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def construct_success(value: _S) -> Success[_S]:
    """Create a [trcks.Success][] object from a value.

    Args:
        value: Value to be wrapped in a [trcks.Success][] object.

    Returns:
        [trcks.Success][] object containing the given value.

    Example:
        >>> from trcks.fp.monads import result as r
        >>> r.construct_success(42)
        ('success', 42)
    """
    return "success", value

map_failure(f)

Create function that maps trcks.Failure values to trcks.Failure values.

trcks.Success values are left unchanged.

Parameters:

Returns:

Example 
>>> from trcks.fp.monads import result as r
>>> add_prefix_to_failure = r.map_failure(lambda s: f"Prefix: {s}")
>>> add_prefix_to_failure(("failure", "negative value"))
('failure', 'Prefix: negative value')
>>> add_prefix_to_failure(("success", 25.0))
('success', 25.0)
Source code in src/trcks/fp/monads/result.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def map_failure(
    f: Callable[[_F1], _F2],
) -> Callable[[Result[_F1, _S1]], Result[_F2, _S1]]:
    """Create function that maps [trcks.Failure][] values to [trcks.Failure][] values.

    [trcks.Success][] values are left unchanged.

    Args:
        f: Function to apply to the [trcks.Failure][] values.

    Returns:
        Maps [trcks.Failure][] values to new [trcks.Failure][] values
            according to the given function and
            leaves [trcks.Success][] values unchanged.

    Example:
        >>> from trcks.fp.monads import result as r
        >>> add_prefix_to_failure = r.map_failure(lambda s: f"Prefix: {s}")
        >>> add_prefix_to_failure(("failure", "negative value"))
        ('failure', 'Prefix: negative value')
        >>> add_prefix_to_failure(("success", 25.0))
        ('success', 25.0)
    """

    def composed_f(value: _F1) -> Failure[_F2]:
        return construct_failure(f(value))

    return map_failure_to_result(composed_f)

map_failure_to_result(f)

Create function that maps trcks.Failure values to trcks.Failure and trcks.Success values.

trcks.Success values are left unchanged.

Parameters:

Returns:

Example 
>>> from trcks.fp.monads import result as r
>>> replace_not_found_failure_by_default_value = r.map_failure_to_result(
...     lambda s: ("success", 0.0) if s == "not found" else ("failure", s)
... )
>>> replace_not_found_failure_by_default_value(("failure", "not found"))
('success', 0.0)
>>> replace_not_found_failure_by_default_value(("failure", "other failure"))
('failure', 'other failure')
>>> replace_not_found_failure_by_default_value(("success", 25.0))
('success', 25.0)
Source code in src/trcks/fp/monads/result.py 
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
193
194
195
196
def map_failure_to_result(
    f: Callable[[_F1], Result[_F2, _S2]],
) -> Callable[[Result[_F1, _S1]], Result[_F2, _S1 | _S2]]:
    """Create function that maps [trcks.Failure][] values
    to [trcks.Failure][] and [trcks.Success][] values.

    [trcks.Success][] values are left unchanged.

    Args:
        f: Function to apply to the [trcks.Failure][] values.

    Returns:
        Maps [trcks.Failure][] values to [trcks.Failure][] and [trcks.Success][] values
            according to the given function and
            leaves [trcks.Success][] values unchanged.

    Example:
        >>> from trcks.fp.monads import result as r
        >>> replace_not_found_failure_by_default_value = r.map_failure_to_result(
        ...     lambda s: ("success", 0.0) if s == "not found" else ("failure", s)
        ... )
        >>> replace_not_found_failure_by_default_value(("failure", "not found"))
        ('success', 0.0)
        >>> replace_not_found_failure_by_default_value(("failure", "other failure"))
        ('failure', 'other failure')
        >>> replace_not_found_failure_by_default_value(("success", 25.0))
        ('success', 25.0)
    """

    def mapped_f(rslt: Result[_F1, _S1]) -> Result[_F2, _S1 | _S2]:
        if rslt[0] == "failure":
            return f(rslt[1])
        if rslt[0] == "success":
            return rslt
        return assert_never(rslt)  # type: ignore [unreachable]  # pragma: no cover

    return mapped_f

map_success(f)

Create function that maps trcks.Success values to trcks.Success values.

trcks.Failure values are left unchanged.

Parameters:

Returns:

Example 
>>> from trcks.fp.monads import result as r
>>> def increase(n: int) -> int:
...     return n + 1
...
>>> increase_success = r.map_success(increase)
>>> increase_success(("failure", "not found"))
('failure', 'not found')
>>> increase_success(("success", 42))
('success', 43)
Source code in src/trcks/fp/monads/result.py 
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
def map_success(
    f: Callable[[_S1], _S2],
) -> Callable[[Result[_F1, _S1]], Result[_F1, _S2]]:
    """Create function that maps [trcks.Success][] values to [trcks.Success][] values.

    [trcks.Failure][] values are left unchanged.

    Args:
        f: Function to apply to the [trcks.Success][] value.

    Returns:
        Leaves [trcks.Failure][] values unchanged and
            maps [trcks.Success][] values to new [trcks.Success][] values
            according to the given function.

    Example:
        >>> from trcks.fp.monads import result as r
        >>> def increase(n: int) -> int:
        ...     return n + 1
        ...
        >>> increase_success = r.map_success(increase)
        >>> increase_success(("failure", "not found"))
        ('failure', 'not found')
        >>> increase_success(("success", 42))
        ('success', 43)
    """

    def composed_f(value: _S1) -> Success[_S2]:
        return construct_success(f(value))

    return map_success_to_result(composed_f)

map_success_to_result(f)

Create function that maps trcks.Success values to trcks.Failure and trcks.Success values.

trcks.Failure values are left unchanged.

Parameters:

Returns:

Example 
>>> import math
>>> from trcks import Result
>>> from trcks.fp.monads import result as r
>>> def _get_square_root(x: float) -> Result[str, float]:
...     if x < 0:
...         return "failure", "negative value"
...     return "success", math.sqrt(x)
...
>>> get_square_root = r.map_success_to_result(_get_square_root)
>>> get_square_root(("failure", "not found"))
('failure', 'not found')
>>> get_square_root(("success", -25.0))
('failure', 'negative value')
>>> get_square_root(("success", 25.0))
('success', 5.0)
Source code in src/trcks/fp/monads/result.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def map_success_to_result(
    f: Callable[[_S1], Result[_F2, _S2]],
) -> Callable[[Result[_F1, _S1]], Result[_F1 | _F2, _S2]]:
    """Create function that maps [trcks.Success][] values
    to [trcks.Failure][] and [trcks.Success][] values.

    [trcks.Failure][] values are left unchanged.

    Args:
        f: Function to apply to the [trcks.Success][] value.

    Returns:
        Leaves [trcks.Failure][] values unchanged and
            maps [trcks.Success][] values to [trcks.Failure][] and
            [trcks.Success][] values according to the given function.

    Example:
        >>> import math
        >>> from trcks import Result
        >>> from trcks.fp.monads import result as r
        >>> def _get_square_root(x: float) -> Result[str, float]:
        ...     if x < 0:
        ...         return "failure", "negative value"
        ...     return "success", math.sqrt(x)
        ...
        >>> get_square_root = r.map_success_to_result(_get_square_root)
        >>> get_square_root(("failure", "not found"))
        ('failure', 'not found')
        >>> get_square_root(("success", -25.0))
        ('failure', 'negative value')
        >>> get_square_root(("success", 25.0))
        ('success', 5.0)
    """

    def mapped_f(rslt: Result[_F1, _S1]) -> Result[_F1 | _F2, _S2]:
        if rslt[0] == "failure":
            return rslt
        if rslt[0] == "success":
            return f(rslt[1])
        return assert_never(rslt)  # type: ignore [unreachable]  # pragma: no cover

    return mapped_f

tap_failure(f)

Create function that applies a side effect to trcks.Failure values.

trcks.Success values are passed on without side effects.

Parameters:

Returns:

Source code in src/trcks/fp/monads/result.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def tap_failure(
    f: Callable[[_F1], object],
) -> Callable[[Result[_F1, _S1]], Result[_F1, _S1]]:
    """Create function that applies a side effect to [trcks.Failure][] values.

    [trcks.Success][] values are passed on without side effects.

    Args:
        f: Side effect to apply to the [trcks.Failure][] value.

    Returns:
        Applies the given side effect to [trcks.Failure][] values and
            returns the original [trcks.Failure][] value.
            Passes on [trcks.Success][] values without side effects.
    """
    return map_failure(i.tap(f))

tap_failure_to_result(f)

Create function that applies a side effect with return type trcks.Result to trcks.Failure values.

trcks.Success values are passed on without side effects.

Parameters:

Returns:

Source code in src/trcks/fp/monads/result.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def tap_failure_to_result(
    f: Callable[[_F1], Result[object, _S2]],
) -> Callable[[Result[_F1, _S1]], Result[_F1, _S1 | _S2]]:
    """Create function that applies a side effect with return type [trcks.Result][]
    to [trcks.Failure][] values.

    [trcks.Success][] values are passed on without side effects.

    Args:
        f: Side effect to apply to the [trcks.Failure][] value.

    Returns:
        Applies the given side effect to [trcks.Failure][] values.
            If the given side effect returns a [trcks.Failure][],
            *the original* [trcks.Failure][] value is returned.
            If the given side effect returns a [trcks.Success][],
            *this* [trcks.Success][] is returned.
            Passes on [trcks.Success][] values without side effects.
    """

    def bypassed_f(value: _F1) -> Result[_F1, _S2]:
        rslt: Result[object, _S2] = f(value)
        if rslt[0] == "failure":
            return construct_failure(value)
        if rslt[0] == "success":
            return rslt
        return assert_never(rslt)  # type: ignore [unreachable]  # pragma: no cover

    return map_failure_to_result(bypassed_f)

tap_success(f)

Create function that applies a side effect to trcks.Success values.

trcks.Failure values are passed on without side effects.

Parameters:

Returns:

Source code in src/trcks/fp/monads/result.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def tap_success(
    f: Callable[[_S1], object],
) -> Callable[[Result[_F1, _S1]], Result[_F1, _S1]]:
    """Create function that applies a side effect to [trcks.Success][] values.

    [trcks.Failure][] values are passed on without side effects.

    Args:
        f: Side effect to apply to the [trcks.Success][] value.

    Returns:
        Passes on [trcks.Failure][] values without side effects.
            Applies the given side effect to [trcks.Success][] values and
            returns the original [trcks.Success][] value.
    """
    return map_success(i.tap(f))

tap_success_to_result(f)

Create function that applies a side effect with return type trcks.Result to trcks.Success values.

trcks.Failure values are passed on without side effects.

Parameters:

Returns:

Source code in src/trcks/fp/monads/result.py 
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
def tap_success_to_result(
    f: Callable[[_S1], Result[_F2, object]],
) -> Callable[[Result[_F1, _S1]], Result[_F1 | _F2, _S1]]:
    """Create function that applies a side effect with return type [trcks.Result][]
    to [trcks.Success][] values.

    [trcks.Failure][] values are passed on without side effects.

    Args:
        f: Side effect to apply to the [trcks.Success][] value.

    Returns:
        Passes on [trcks.Failure][] values without side effects.
            Applies the given side effect to [trcks.Success][] values.
            If the given side effect returns a [trcks.Failure][],
            *this* [trcks.Failure][] is returned.
            If the given side effect returns a [trcks.Success][],
            *the original* [trcks.Success][] value is returned.
    """

    def bypassed_f(value: _S1) -> Result[_F2, _S1]:
        rslt: Result[_F2, object] = f(value)
        if rslt[0] == "failure":
            return rslt
        if rslt[0] == "success":
            return construct_success(value)
        return assert_never(rslt)  # type: ignore [unreachable]  # pragma: no cover

    return map_success_to_result(bypassed_f)