Railway-oriented programming with trcks.fp

Railway-oriented programming (ROP) is a pattern for composing functions that may fail, keeping error handling clean and explicit. This page covers the functional style: you build a pipeline as a tuple of a start value and a sequence of functions, then run it with trcks.fp.composition.pipe — a "build a pipeline, then run it" mental model. Single-track and double-track code are both discussed. Synchronous and asynchronous code are also discussed.

Prerequisites

Read Tuple types provided by trcks first; this page assumes familiarity with trcks.Failure, trcks.Success, and trcks.Result.
The motivation page for ROP explains the design rationale.
The async sections assume comfort with async, await, and asyncio.run.
See the glossary for definitions of terms such as "single-track", "double-track", and "short-circuiting".
See the OOP and FP equivalence table for a side-by-side comparison with the object-oriented style.

Synchronous single-track code for a single value with trcks.fp.composition

The function trcks.fp.composition.pipe allows us to chain functions:

Example

>>> from trcks.fp.composition import pipe
>>>
>>> def to_length_string(s: str) -> str:
...     return pipe((s, len, lambda n: f"Length: {n}"))
...
>>> to_length_string("Hello, world!")
'Length: 13'

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> pipe(("Hello, world!",))
'Hello, world!'
>>> pipe(("Hello, world!", len))
13
>>> pipe(("Hello, world!", len, lambda n: f"Length: {n}"))
'Length: 13'

Note

The function trcks.fp.composition.pipe expects a trcks.fp.composition.Pipeline, i.e. a tuple consisting of a start value followed by up to seven compatible functions.

Side effects like logging or writing to a file tend to "consume" their input and return None instead. To avoid this, we can use the higher-order function trcks.fp.monads.identity.tap. This higher-order function turns each function into a function that behaves like the original function but returns the input value.

Example

>>> from trcks.fp.monads import identity as i
>>>
>>> def to_length_string(s: str) -> str:
...     return pipe(
...         (
...             s,
...             i.tap(lambda o: print(f"LOG: Received '{o}'.")),
...             len,
...             lambda n: f"Length: {n}",
...             i.tap(lambda o: print(f"LOG: Returning '{o}'.")),
...         ),
...     )
...
>>> output = to_length_string("Hello, world!")
LOG: Received 'Hello, world!'.
LOG: Returning 'Length: 13'.
>>> output
'Length: 13'

Synchronous double-track code for a single value with trcks.fp.composition and trcks.fp.monads.result

If one of the functions in a trcks.fp.composition.Pipeline returns a trcks.Result[F, S] type, the following function must accept this trcks.Result[F, S] type as its input. However, functions with input type trcks.Result[F, S] tend to violate the "do one thing and do it well" principle. Therefore, the module trcks.fp.monads.result provides some higher-order functions named map_* that turn functions with input type F and functions with input type S into functions with input type trcks.Result[F, S].

Example

>>> from typing import Literal
>>> from trcks import Result
>>> from trcks.fp.monads import result as r
>>>
>>> UserDoesNotExist = Literal["User does not exist"]
>>> UserDoesNotHaveASubscription = Literal["User does not have a subscription"]
>>> FailureDescription = UserDoesNotExist | UserDoesNotHaveASubscription
>>>
>>> def get_user_id(user_email: str) -> Result[UserDoesNotExist, int]:
...     if user_email == "erika.mustermann@domain.org":
...         return "success", 1
...     if user_email == "john_doe@provider.com":
...         return "success", 2
...     return "failure", "User does not exist"
...
>>> def get_subscription_id(
...     user_id: int
... ) -> Result[UserDoesNotHaveASubscription, int]:
...     if user_id == 1:
...         return "success", 42
...     return "failure", "User does not have a subscription"
...
>>> def get_subscription_fee(subscription_id: int) -> float:
...     return subscription_id * 0.1
...
>>> def get_subscription_fee_by_email(
...     user_email: str
... ) -> Result[FailureDescription, float]:
...     # Explicitly assigning a type to `pipeline` might
...     # help your static type checker understand that
...     # `pipeline` is a valid argument for `pipe`:
...     pipeline: Pipeline3[
...         str,
...         Result[UserDoesNotExist, int],
...         Result[FailureDescription, int],
...         Result[FailureDescription, float],
...     ] = (
...         user_email,
...         get_user_id,
...         r.map_success_to_result(get_subscription_id),
...         r.map_success(get_subscription_fee),
...     )
...     return pipe(pipeline)
...
>>> get_subscription_fee_by_email("erika.mustermann@domain.org")
('success', 4.2)
>>> get_subscription_fee_by_email("john_doe@provider.com")
('failure', 'User does not have a subscription')
>>> get_subscription_fee_by_email("jane_doe@provider.com")
('failure', 'User does not exist')

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> from trcks.fp.composition import (
...     Pipeline0, Pipeline1, Pipeline2, Pipeline3, pipe
... )
>>>
>>> p0: Pipeline0[str] = ("erika.mustermann@domain.org",)
>>> pipe(p0)
'erika.mustermann@domain.org'
>>>
>>> p1: Pipeline1[str, Result[UserDoesNotExist, int]] = (
...     "erika.mustermann@domain.org",
...     get_user_id,
... )
>>> pipe(p1)
('success', 1)
>>>
>>> p2: Pipeline2[
...     str, Result[UserDoesNotExist, int], Result[FailureDescription, int]
... ] = (
...     "erika.mustermann@domain.org",
...     get_user_id,
...     r.map_success_to_result(get_subscription_id),
... )
>>> pipe(p2)
('success', 42)
>>>
>>> p3: Pipeline3[
...     str,
...     Result[UserDoesNotExist, int],
...     Result[FailureDescription, int],
...     Result[FailureDescription, float],
... ] = (
...     "erika.mustermann@domain.org",
...     get_user_id,
...     r.map_success_to_result(get_subscription_id),
...     r.map_success(get_subscription_fee),
... )
>>> pipe(p3)
('success', 4.2)

While trcks.fp.monads.result.map_failure and trcks.fp.monads.result.map_success allow us to apply functions in the failure case or in the success case, respectively, the higher-order functions trcks.fp.monads.result.tap_failure and trcks.fp.monads.result.tap_success allow us to execute side effects in the failure case or in the success case, respectively.

Example

>>> from trcks.fp.composition import Pipeline6
>>>
>>> def get_subscription_fee_by_email(
...     user_email: str
... ) -> Result[FailureDescription, float]:
...     pipeline: Pipeline6[
...         str,
...         Result[UserDoesNotExist, int],
...         Result[UserDoesNotExist, int],
...         Result[FailureDescription, int],
...         Result[FailureDescription, float],
...         Result[FailureDescription, float],
...         Result[FailureDescription, float],
...     ] = (
...         user_email,
...         get_user_id,
...         r.tap_success(lambda n: print(f"LOG: User ID: {n}.")),
...         r.map_success_to_result(get_subscription_id),
...         r.map_success(get_subscription_fee),
...         r.tap_success(lambda x: print(f"LOG: Subscription fee: {x}.")),
...         r.tap_failure(lambda fd: print(f"LOG: Failure description: {fd}.")),
...     )
...     return pipe(pipeline)
...
>>> fee_erika = get_subscription_fee_by_email("erika.mustermann@domain.org")
LOG: User ID: 1.
LOG: Subscription fee: 4.2.
>>> fee_erika
('success', 4.2)
>>>
>>> fee_john = get_subscription_fee_by_email("john_doe@provider.com")
LOG: User ID: 2.
LOG: Failure description: User does not have a subscription.
>>> fee_john
('failure', 'User does not have a subscription')
>>>
>>> fee_jane = get_subscription_fee_by_email("jane_doe@provider.com")
LOG: Failure description: User does not exist.
>>> fee_jane
('failure', 'User does not exist')

Sometimes, side effects themselves can fail and need to return a trcks.Result type. The higher-order function trcks.fp.monads.result.tap_success_to_result allows us to execute such side effects in the success case. If the side effect returns a trcks.Failure, that failure is propagated. If the side effect returns a trcks.Success, the original success value is preserved.

Example

>>> OutOfDiskSpace = Literal["Out of disk space"]
>>>
>>> def write_to_disk(n: int) -> Result[OutOfDiskSpace, None]:
...     if n > 1:
...         return "failure", "Out of disk space"
...     return "success", print(f"LOG: Wrote {n} to disk.")
...
>>> def get_and_persist_user_id(
...     user_email: str
... ) -> Result[UserDoesNotExist | OutOfDiskSpace, int]:
...     pipeline: Pipeline2[
...         str,
...         Result[UserDoesNotExist, int],
...         Result[UserDoesNotExist | OutOfDiskSpace, int],
...     ] = (
...         user_email,
...         get_user_id,
...         r.tap_success_to_result(write_to_disk),
...     )
...     return pipe(pipeline)
...
>>> id_erika = get_and_persist_user_id("erika.mustermann@domain.org")
LOG: Wrote 1 to disk.
>>> id_erika
('success', 1)
>>>
>>> id_john = get_and_persist_user_id("john_doe@provider.com")
>>> id_john
('failure', 'Out of disk space')
>>>
>>> id_jane = get_and_persist_user_id("jane_doe@provider.com")
>>> id_jane
('failure', 'User does not exist')

Asynchronous single-track code for a single value with trcks.fp.composition and trcks.fp.monads.awaitable

If one of the functions in a trcks.fp.composition.Pipeline returns a collections.abc.Awaitable[T] type, the following function must accept this collections.abc.Awaitable[T] type as its input. However, functions with input type collections.abc.Awaitable[T] tend to contain unnecessary await statements. Therefore, the module trcks.fp.monads.awaitable provides some higher-order functions named map_* that turn functions with input type T into functions with input type collections.abc.Awaitable[T].

Example

>>> import asyncio
>>> from collections.abc import Awaitable
>>> from trcks.fp.monads import awaitable as a
>>>
>>> async def read_from_disk(path: str) -> str:
...     await asyncio.sleep(0.001)
...     s = "Hello, world!"
...     print(f"Read '{s}' from file {path}.")
...     return s
...
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> async def write_to_disk(s: str, path: str) -> None:
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' to file {path}.")
...
>>> async def read_and_transform_and_write(
...     input_path: str, output_path: str
... ) -> None:
...     p: Pipeline3[
...         str, Awaitable[str], Awaitable[str], Awaitable[None]
...     ] = (
...         input_path,
...         read_from_disk,
...         a.map_(transform),
...         a.map_to_awaitable(lambda s: write_to_disk(s, output_path)),
...     )
...     return await pipe(p)
...
>>> asyncio.run(read_and_transform_and_write("input.txt", "output.txt"))
Read 'Hello, world!' from file input.txt.
Wrote 'Length: 13' to file output.txt.

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> p1: Pipeline1[str, Awaitable[str]] = (
...     "input.txt",
...     read_from_disk,
... )
>>> asyncio.run(a.to_coroutine(pipe(p1)))
Read 'Hello, world!' from file input.txt.
'Hello, world!'
>>>
>>> p2: Pipeline2[str, Awaitable[str], Awaitable[str]] = (
...     "input.txt",
...     read_from_disk,
...     a.map_(transform),
... )
>>> asyncio.run(a.to_coroutine(pipe(p2)))
Read 'Hello, world!' from file input.txt.
'Length: 13'
>>>
>>> p3: Pipeline3[str, Awaitable[str], Awaitable[str], Awaitable[None]] = (
...     "input.txt",
...     read_from_disk,
...     a.map_(transform),
...     a.map_to_awaitable(lambda s: write_to_disk(s, "output.txt")),
... )
>>> asyncio.run(a.to_coroutine(pipe(p3)))
Read 'Hello, world!' from file input.txt.
Wrote 'Length: 13' to file output.txt.

Note

The values pipe(p1), pipe(p2), and pipe(p3) are all of the type collections.abc.Awaitable. Since asyncio.run expects the input type collections.abc.Coroutine, we use the function trcks.fp.monads.awaitable.to_coroutine to convert the collections.abc.Awaitables to collections.abc.Coroutines.

The higher-order function trcks.fp.monads.awaitable.tap allows us to execute synchronous side effects. Similarly, the higher-order function trcks.fp.monads.awaitable.tap_to_awaitable allows us to execute asynchronous side effects.

Example

>>> async def read_from_disk(path: str) -> str:
...     await asyncio.sleep(0.001)
...     return "Hello, world!"
...
>>> async def write_to_disk(s: str, path: str) -> None:
...     await asyncio.sleep(0.001)
...
>>> async def read_and_transform_and_write(
...     input_path: str, output_path: str
... ) -> str:
...     p: Pipeline5[
...         str,
...         Awaitable[str],
...         Awaitable[str],
...         Awaitable[str],
...         Awaitable[str],
...         Awaitable[str],
...     ] = (
...         input_path,
...         read_from_disk,
...         a.tap(lambda s: print(f"Read '{s}' from disk.")),
...         a.map_(transform),
...         a.tap_to_awaitable(lambda s: write_to_disk(s, output_path)),
...         a.tap(lambda s: print(f"Wrote '{s}' to disk.")),
...     )
...     return await pipe(p)
...
>>> asyncio.run(read_and_transform_and_write("input.txt", "output.txt"))
Read 'Hello, world!' from disk.
Wrote 'Length: 13' to disk.
'Length: 13'

Asynchronous double-track code for a single value with trcks.fp.composition and trcks.fp.monads.awaitable_result

If one of the functions in a trcks.fp.composition.Pipeline returns a trcks.AwaitableResult[F, S] type, the following function must accept this trcks.AwaitableResult[F, S] type as its input. However, functions with input type trcks.AwaitableResult[F, S] tend to contain unnecessary await statements and violate the "do one thing and do it well" principle. Therefore, the module trcks.fp.monads.awaitable_result provides some higher-order functions named map_* that turn functions with input type F and functions with input type S into functions with input type trcks.AwaitableResult[F, S].

Example

>>> from trcks.fp.monads import awaitable_result as ar
>>>
>>> ReadErrorLiteral = Literal["read error"]
>>> WriteErrorLiteral = Literal["write error"]
>>>
>>> async def read_from_disk(path: str) -> Result[ReadErrorLiteral, str]:
...     if path != "input.txt":
...         return "failure", "read error"
...     await asyncio.sleep(0.001)
...     s = "Hello, world!"
...     print(f"Read '{s}' from file {path}.")
...     return "success", s
...
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> async def write_to_disk(
...     s: str, path: str
... ) -> Result[WriteErrorLiteral, None]:
...     if path != "output.txt":
...         return "failure", "write error"
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' to file {path}.")
...     return "success", None
...
>>> async def read_and_transform_and_write(
...     input_path: str, output_path: str
... ) -> Result[ReadErrorLiteral | WriteErrorLiteral, None]:
...     p: Pipeline3[
...         str,
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral | WriteErrorLiteral, None],
...     ] = (
...         input_path,
...         read_from_disk,
...         ar.map_success(transform),
...         ar.map_success_to_awaitable_result(lambda s: write_to_disk(s, output_path)),
...     )
...     return await pipe(p)
...
>>> asyncio.run(read_and_transform_and_write("input.txt", "output.txt"))
Read 'Hello, world!' from file input.txt.
Wrote 'Length: 13' to file output.txt.
('success', None)

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> from trcks import AwaitableResult, Result
>>>
>>> p1: Pipeline1[str, AwaitableResult[ReadErrorLiteral, str]] = (
...     "input.txt",
...     read_from_disk,
... )
>>> asyncio.run(ar.to_coroutine_result(pipe(p1)))
Read 'Hello, world!' from file input.txt.
('success', 'Hello, world!')
>>>
>>> p2: Pipeline2[
...     str,
...     AwaitableResult[ReadErrorLiteral, str],
...     AwaitableResult[ReadErrorLiteral, str],
... ] = (
...     "input.txt",
...     read_from_disk,
...     ar.map_success(transform),
... )
>>> asyncio.run(ar.to_coroutine_result(pipe(p2)))
Read 'Hello, world!' from file input.txt.
('success', 'Length: 13')
>>>
>>> p3: Pipeline3[
...     str,
...     AwaitableResult[ReadErrorLiteral, str],
...     AwaitableResult[ReadErrorLiteral, str],
...     AwaitableResult[ReadErrorLiteral | WriteErrorLiteral, None],
... ] = (
...     "input.txt",
...     read_from_disk,
...     ar.map_success(transform),
...     ar.map_success_to_awaitable_result(lambda s: write_to_disk(s, "output.txt")),
... )
>>> asyncio.run(ar.to_coroutine_result(pipe(p3)))
Read 'Hello, world!' from file input.txt.
Wrote 'Length: 13' to file output.txt.
('success', None)

Note

The values pipe(p1), pipe(p2), and pipe(p3) are all of type trcks.AwaitableResult. Since asyncio.run expects the input type collections.abc.Coroutine, we use the function trcks.fp.monads.awaitable_result.to_coroutine_result to convert the trcks.AwaitableResults to collections.abc.Coroutines.

The higher-order functions trcks.fp.monads.awaitable_result.tap_failure and trcks.fp.monads.awaitable_result.tap_success allow us to execute synchronous side effects in the failure case or in the success case, respectively:

Example

>>> async def read_from_disk(path: str) -> Result[ReadErrorLiteral, str]:
...     if path != "input.txt":
...         return "failure", "read error"
...     await asyncio.sleep(0.001)
...     return "success", "Hello, world!"
...
>>> async def write_to_disk(
...     s: str, path: str
... ) -> Result[WriteErrorLiteral, None]:
...     if path != "output.txt":
...         return "failure", "write error"
...     await asyncio.sleep(0.001)
...     return "success", None
...
>>> async def read_and_transform_and_write(
...     input_path: str, output_path: str
... ) -> Result[ReadErrorLiteral | WriteErrorLiteral, None]:
...     pipeline: Pipeline6[
...         str,
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral | WriteErrorLiteral, None],
...         AwaitableResult[ReadErrorLiteral | WriteErrorLiteral, None],
...         AwaitableResult[ReadErrorLiteral | WriteErrorLiteral, None],
...     ] = (
...         input_path,
...         read_from_disk,
...         ar.tap_success(lambda s: print(f"LOG: Read '{s}' from disk.")),
...         ar.map_success(transform),
...         ar.map_success_to_awaitable_result(lambda s: write_to_disk(s, output_path)),
...         ar.tap_success(lambda _: print("LOG: Successfully wrote to disk.")),
...         ar.tap_failure(lambda err: print(f"LOG: Failed with error: {err}")),
...     )
...     return await pipe(pipeline)
...
>>> result_1 = asyncio.run(read_and_transform_and_write("input.txt", "output.txt"))
LOG: Read 'Hello, world!' from disk.
LOG: Successfully wrote to disk.
>>> result_1
('success', None)
>>>
>>> result_2 = asyncio.run(read_and_transform_and_write("missing.txt", "output.txt"))
LOG: Failed with error: read error
>>> result_2
('failure', 'read error')

Sometimes, side effects themselves can fail and need to return an trcks.AwaitableResult type. The higher-order function trcks.fp.monads.awaitable_result.tap_success_to_awaitable_result allows us to execute such asynchronous side effects in the success case. If the side effect returns an trcks.AwaitableFailure, that failure is propagated. If the side effect returns an trcks.AwaitableSuccess, the original success value is preserved:

Example

>>> async def write_to_disk(s: str) -> Result[OutOfDiskSpace, None]:
...     await asyncio.sleep(0.001)
...     if len(s) > 10:
...         return "failure", "Out of disk space"
...     return "success", None
...
>>> async def read_and_persist(
...     input_path: str
... ) -> Result[ReadErrorLiteral | OutOfDiskSpace, str]:
...     pipeline: Pipeline3[
...         str,
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral, str],
...         AwaitableResult[ReadErrorLiteral | OutOfDiskSpace, str],
...     ] = (
...         input_path,
...         read_from_disk,
...         ar.tap_success(lambda s: print(f"LOG: Persisting '{s}'.")),
...         ar.tap_success_to_awaitable_result(write_to_disk),
...     )
...     return await pipe(pipeline)
...
>>> result = asyncio.run(read_and_persist("input.txt"))
LOG: Persisting 'Hello, world!'.
>>> result
('failure', 'Out of disk space')

Synchronous single-track code for a homogeneous tuple with trcks.fp.composition and trcks.fp.monads.tuple_

If we want to apply a pipeline of functions to each element in a tuple, the module trcks.fp.monads.tuple_ provides some higher-order functions named map_* and tap that turn element-wise functions into functions operating on entire tuples.

Example

>>> from trcks.fp.monads import tuple_ as t
>>>
>>> def normalize_email(email: str) -> str:
...     return email.strip().lower()
...
>>> def to_domain(email: str) -> str:
...     return email.split("@")[1]
...
>>> def get_domains(emails: tuple[str, ...]) -> tuple[str, ...]:
...     pipeline: Pipeline2[
...         tuple[str, ...],
...         tuple[str, ...],
...         tuple[str, ...],
...     ] = (
...         emails,
...         t.map_(normalize_email),
...         t.map_(to_domain),
...     )
...     return pipe(pipeline)
...
>>> get_domains(("  Erika.Mustermann@Domain.ORG ", "JOHN_DOE@Provider.COM  "))
('domain.org', 'provider.com')

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> emails: tuple[str, ...] = (
...     "  Erika.Mustermann@Domain.ORG ",
...     "JOHN_DOE@Provider.COM  ",
... )
>>>
>>> p0: Pipeline0[tuple[str, ...]] = (emails,)
>>> pipe(p0)
('  Erika.Mustermann@Domain.ORG ', 'JOHN_DOE@Provider.COM  ')
>>>
>>> p1: Pipeline1[tuple[str, ...], tuple[str, ...]] = (
...     emails,
...     t.map_(normalize_email),
... )
>>> pipe(p1)
('erika.mustermann@domain.org', 'john_doe@provider.com')
>>>
>>> p2: Pipeline2[tuple[str, ...], tuple[str, ...], tuple[str, ...]] = (
...     emails,
...     t.map_(normalize_email),
...     t.map_(to_domain),
... )
>>> pipe(p2)
('domain.org', 'provider.com')

Note

The higher-order function trcks.fp.monads.tuple_.map_to_iterable applies a function that returns a collections.abc.Iterable to each element and flattens the results (like a "flat map"):

>>> pipe(
...     (
...         ("ab", "cd"),
...         t.map_to_iterable(tuple),
...     )
... )
('a', 'b', 'c', 'd')

The higher-order function trcks.fp.monads.tuple_.tap allows us to execute side effects for each element:

Example

>>> def get_domains(emails: tuple[str, ...]) -> tuple[str, ...]:
...     pipeline: Pipeline3[
...         tuple[str, ...],
...         tuple[str, ...],
...         tuple[str, ...],
...         tuple[str, ...],
...     ] = (
...         emails,
...         t.map_(normalize_email),
...         t.tap(lambda e: print(f"LOG: Processing '{e}'.")),
...         t.map_(to_domain),
...     )
...     return pipe(pipeline)
...
>>> result = get_domains(
...     ("  Erika.Mustermann@Domain.ORG ", "JOHN_DOE@Provider.COM  ")
... )
LOG: Processing 'erika.mustermann@domain.org'.
LOG: Processing 'john_doe@provider.com'.
>>> result
('domain.org', 'provider.com')

Synchronous double-track code for a homogeneous tuple with trcks.fp.composition and trcks.fp.monads.result_tuple

If one of the functions in a trcks.fp.composition.Pipeline returns a trcks.ResultTuple[F, S] type, the module trcks.fp.monads.result_tuple provides some higher-order functions named map_successes* and tap_successes* that turn element-wise functions into functions operating on trcks.ResultTuple values. The success track functions use the plural map_successes (instead of map_success) because they operate on each element in the trcks.SuccessTuple individually. Processing short-circuits on the first trcks.Failure.

Example

>>> from trcks import ResultTuple, SuccessTuple
>>> from trcks.fp.composition import Pipeline4
>>> from trcks.fp.monads import result_tuple as rt
>>>
>>> UserDoesNotExist = Literal["User does not exist"]
>>> UserDoesNotHaveASubscription = Literal["User does not have a subscription"]
>>> FailureDescription = UserDoesNotExist | UserDoesNotHaveASubscription
>>>
>>> def get_user_id(user_email: str) -> Result[UserDoesNotExist, int]:
...     if user_email == "erika.mustermann@domain.org":
...         return "success", 1
...     if user_email == "john_doe@provider.com":
...         return "success", 2
...     return "failure", "User does not exist"
...
>>> def get_subscription_id(
...     user_id: int
... ) -> Result[UserDoesNotHaveASubscription, int]:
...     if user_id == 1:
...         return "success", 42
...     return "failure", "User does not have a subscription"
...
>>> def get_subscription_fee(subscription_id: int) -> float:
...     return subscription_id * 0.1
...
>>> def get_subscription_fees_by_email(
...     user_emails: tuple[str, ...],
... ) -> ResultTuple[FailureDescription, float]:
...     pipeline: Pipeline4[
...         tuple[str, ...],
...         SuccessTuple[str],
...         ResultTuple[UserDoesNotExist, int],
...         ResultTuple[FailureDescription, int],
...         ResultTuple[FailureDescription, float],
...     ] = (
...         user_emails,
...         rt.construct_successes_from_iterable,
...         rt.map_successes_to_result(get_user_id),
...         rt.map_successes_to_result(get_subscription_id),
...         rt.map_successes(get_subscription_fee),
...     )
...     return pipe(pipeline)
...
>>> get_subscription_fees_by_email(("erika.mustermann@domain.org",))
('success', (4.2,))
>>> get_subscription_fees_by_email(
...     ("erika.mustermann@domain.org", "john_doe@provider.com")
... )
('failure', 'User does not have a subscription')
>>> get_subscription_fees_by_email(("jane_doe@provider.com",))
('failure', 'User does not exist')

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> from trcks.fp.composition import Pipeline4
>>>
>>> p0: Pipeline0[tuple[str, ...]] = (("erika.mustermann@domain.org",),)
>>> pipe(p0)
('erika.mustermann@domain.org',)
>>>
>>> p1: Pipeline1[
...     tuple[str, ...],
...     SuccessTuple[str],
... ] = (
...     ("erika.mustermann@domain.org",),
...     rt.construct_successes_from_iterable,
... )
>>> pipe(p1)
('success', ('erika.mustermann@domain.org',))
>>>
>>> p2: Pipeline2[
...     tuple[str, ...],
...     SuccessTuple[str],
...     ResultTuple[UserDoesNotExist, int],
... ] = (
...     ("erika.mustermann@domain.org",),
...     rt.construct_successes_from_iterable,
...     rt.map_successes_to_result(get_user_id),
... )
>>> pipe(p2)
('success', (1,))
>>>
>>> p3: Pipeline3[
...     tuple[str, ...],
...     SuccessTuple[str],
...     ResultTuple[UserDoesNotExist, int],
...     ResultTuple[FailureDescription, int],
... ] = (
...     ("erika.mustermann@domain.org",),
...     rt.construct_successes_from_iterable,
...     rt.map_successes_to_result(get_user_id),
...     rt.map_successes_to_result(get_subscription_id),
... )
>>> pipe(p3)
('success', (42,))
>>>
>>> p4: Pipeline4[
...     tuple[str, ...],
...     SuccessTuple[str],
...     ResultTuple[UserDoesNotExist, int],
...     ResultTuple[FailureDescription, int],
...     ResultTuple[FailureDescription, float],
... ] = (
...     ("erika.mustermann@domain.org",),
...     rt.construct_successes_from_iterable,
...     rt.map_successes_to_result(get_user_id),
...     rt.map_successes_to_result(get_subscription_id),
...     rt.map_successes(get_subscription_fee),
... )
>>> pipe(p4)
('success', (4.2,))

Note

The function trcks.fp.monads.result_tuple.construct_successes_from_iterable converts a collections.abc.Iterable into a tuple and wraps it into a trcks.SuccessTuple, which can then be used with the higher-order functions from trcks.fp.monads.result_tuple.

While trcks.fp.monads.result.map_failure and trcks.fp.monads.result.map_success operate on single-value trcks.Result types, trcks.fp.monads.result_tuple.map_successes (plural) operates on each element in the tuple.

The higher-order functions trcks.fp.monads.result_tuple.tap_successes and trcks.fp.monads.result_tuple.tap_failure allow us to execute side effects in the success case (for each element) or in the failure case, respectively.

Example

>>> from trcks.fp.composition import Pipeline7
>>> def get_subscription_fees_by_email(
...     user_emails: tuple[str, ...],
... ) -> ResultTuple[FailureDescription, float]:
...     pipeline: Pipeline7[
...         tuple[str, ...],
...         SuccessTuple[str],
...         ResultTuple[UserDoesNotExist, int],
...         ResultTuple[UserDoesNotExist, int],
...         ResultTuple[FailureDescription, int],
...         ResultTuple[FailureDescription, float],
...         ResultTuple[FailureDescription, float],
...         ResultTuple[FailureDescription, float],
...     ] = (
...         user_emails,
...         rt.construct_successes_from_iterable,
...         rt.map_successes_to_result(get_user_id),
...         rt.tap_successes(lambda n: print(f"LOG: User ID: {n}.")),
...         rt.map_successes_to_result(get_subscription_id),
...         rt.map_successes(get_subscription_fee),
...         rt.tap_successes(lambda x: print(f"LOG: Subscription fee: {x}.")),
...         rt.tap_failure(lambda fd: print(f"LOG: Failure: {fd}.")),
...     )
...     return pipe(pipeline)
...
>>> fees_erika = get_subscription_fees_by_email(
...     ("erika.mustermann@domain.org",)
... )
LOG: User ID: 1.
LOG: Subscription fee: 4.2.
>>> fees_erika
('success', (4.2,))
>>>
>>> fees_john = get_subscription_fees_by_email(
...     ("john_doe@provider.com",)
... )
LOG: User ID: 2.
LOG: Failure: User does not have a subscription.
>>> fees_john
('failure', 'User does not have a subscription')
>>>
>>> fees_jane = get_subscription_fees_by_email(
...     ("jane_doe@provider.com",)
... )
LOG: Failure: User does not exist.
>>> fees_jane
('failure', 'User does not exist')

Sometimes, side effects themselves can fail and need to return a trcks.Result type. The higher-order function trcks.fp.monads.result_tuple.tap_successes_to_result allows us to execute such side effects for each element in the success case. If the side effect returns a trcks.Failure for any element, that failure is propagated. If the side effect returns a trcks.Success for all elements, the original success values are preserved.

Example

>>> OutOfDiskSpace = Literal["Out of disk space"]
>>>
>>> def write_to_disk(n: int) -> Result[OutOfDiskSpace, None]:
...     if n > 1:
...         return "failure", "Out of disk space"
...     return "success", print(f"LOG: Wrote {n} to disk.")
...
>>> def get_and_persist_user_ids(
...     user_emails: tuple[str, ...],
... ) -> ResultTuple[UserDoesNotExist | OutOfDiskSpace, int]:
...     pipeline: Pipeline3[
...         tuple[str, ...],
...         SuccessTuple[str],
...         ResultTuple[UserDoesNotExist, int],
...         ResultTuple[UserDoesNotExist | OutOfDiskSpace, int],
...     ] = (
...         user_emails,
...         rt.construct_successes_from_iterable,
...         rt.map_successes_to_result(get_user_id),
...         rt.tap_successes_to_result(write_to_disk),
...     )
...     return pipe(pipeline)
...
>>> ids_erika = get_and_persist_user_ids(("erika.mustermann@domain.org",))
LOG: Wrote 1 to disk.
>>> ids_erika
('success', (1,))
>>>
>>> ids_john = get_and_persist_user_ids(("john_doe@provider.com",))
>>> ids_john
('failure', 'Out of disk space')
>>>
>>> ids_jane = get_and_persist_user_ids(("jane_doe@provider.com",))
>>> ids_jane
('failure', 'User does not exist')

To demonstrate that processing truly short-circuits on the first trcks.Failure, we can add logging to get_user_id. When the second element fails, the third element is never evaluated:

Example

>>> def get_user_id_logged(user_email: str) -> Result[UserDoesNotExist, int]:
...     print(f"LOG: Looking up '{user_email}'.")
...     return get_user_id(user_email)
...
>>> pipe(
...     (
...         (
...             "erika.mustermann@domain.org",
...             "jane_doe@provider.com",
...             "john_doe@provider.com",
...         ),
...         rt.construct_successes_from_iterable,
...         rt.map_successes_to_result(get_user_id_logged),
...     )
... )
LOG: Looking up 'erika.mustermann@domain.org'.
LOG: Looking up 'jane_doe@provider.com'.
('failure', 'User does not exist')

"john_doe@provider.com" is never looked up: as soon as "jane_doe@provider.com" returns a trcks.Failure, the remaining elements are skipped.

Asynchronous single-track code for a homogeneous tuple with trcks.fp.composition and trcks.fp.monads.awaitable_tuple

If one of the functions in a trcks.fp.composition.Pipeline returns a trcks.AwaitableTuple[T] type, the following function must accept this trcks.AwaitableTuple[T] type as its input. The module trcks.fp.monads.awaitable_tuple provides some higher-order functions named map_* that turn element-wise functions into functions operating on trcks.AwaitableTuple values.

Example

>>> from trcks import AwaitableTuple
>>> from trcks.fp.monads import awaitable_tuple as at
>>>
>>> async def read_from_disk(path: str) -> str:
...     await asyncio.sleep(0.001)
...     contents = {"a.txt": "Hello", "b.txt": "World"}
...     return contents[path]
...
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> async def read_and_transform(
...     input_paths: tuple[str, ...],
... ) -> tuple[str, ...]:
...     p: Pipeline3[
...         tuple[str, ...],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...     ] = (
...         input_paths,
...         at.construct_from_iterable,
...         at.map_to_awaitable(read_from_disk),
...         at.map_(transform),
...     )
...     return await pipe(p)
...
>>> asyncio.run(read_and_transform(("a.txt", "b.txt")))
('Length: 5', 'Length: 5')

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> from trcks import AwaitableTuple
>>>
>>> p1: Pipeline1[tuple[str, ...], AwaitableTuple[str]] = (
...     ("a.txt", "b.txt"),
...     at.construct_from_iterable,
... )
>>> asyncio.run(at.to_coroutine_tuple(pipe(p1)))
('a.txt', 'b.txt')
>>>
>>> p2: Pipeline2[
...     tuple[str, ...],
...     AwaitableTuple[str],
...     AwaitableTuple[str],
... ] = (
...     ("a.txt", "b.txt"),
...     at.construct_from_iterable,
...     at.map_to_awaitable(read_from_disk),
... )
>>> asyncio.run(at.to_coroutine_tuple(pipe(p2)))
('Hello', 'World')
>>>
>>> p3: Pipeline3[
...     tuple[str, ...],
...     AwaitableTuple[str],
...     AwaitableTuple[str],
...     AwaitableTuple[str],
... ] = (
...     ("a.txt", "b.txt"),
...     at.construct_from_iterable,
...     at.map_to_awaitable(read_from_disk),
...     at.map_(transform),
... )
>>> asyncio.run(at.to_coroutine_tuple(pipe(p3)))
('Length: 5', 'Length: 5')

Note

The function trcks.fp.monads.awaitable_tuple.construct_from_iterable converts a collections.abc.Iterable into a tuple and wraps it into a trcks.AwaitableTuple, which can then be used with the higher-order functions from trcks.fp.monads.awaitable_tuple.

The values pipe(p1), pipe(p2), and pipe(p3) are all of type trcks.AwaitableTuple. Since asyncio.run expects the input type collections.abc.Coroutine, we use the function trcks.fp.monads.awaitable_tuple.to_coroutine_tuple to convert the trcks.AwaitableTuples to collections.abc.Coroutines.

The higher-order function trcks.fp.monads.awaitable_tuple.tap allows us to execute synchronous side effects for each element. Similarly, the higher-order function trcks.fp.monads.awaitable_tuple.tap_to_awaitable allows us to execute asynchronous side effects for each element.

Example

>>> from trcks.fp.composition import Pipeline5
>>> async def read_from_disk(path: str) -> str:
...     await asyncio.sleep(0.001)
...     contents = {"a.txt": "Hello", "b.txt": "World"}
...     return contents[path]
...
>>> async def read_and_transform(
...     input_paths: tuple[str, ...],
... ) -> tuple[str, ...]:
...     p: Pipeline5[
...         tuple[str, ...],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...         AwaitableTuple[str],
...     ] = (
...         input_paths,
...         at.construct_from_iterable,
...         at.map_to_awaitable(read_from_disk),
...         at.tap(lambda s: print(f"Read '{s}' from disk.")),
...         at.map_(transform),
...         at.tap(lambda s: print(f"Transformed to '{s}'.")),
...     )
...     return await pipe(p)
...
>>> asyncio.run(read_and_transform(("a.txt", "b.txt")))
Read 'Hello' from disk.
Read 'World' from disk.
Transformed to 'Length: 5'.
Transformed to 'Length: 5'.
('Length: 5', 'Length: 5')

Asynchronous double-track code for a homogeneous tuple with trcks.fp.composition and trcks.fp.monads.awaitable_result_tuple

If one of the functions in a trcks.fp.composition.Pipeline returns a trcks.AwaitableResultTuple[F, S] type, the module trcks.fp.monads.awaitable_result_tuple provides some higher-order functions named map_successes* and tap_successes* that turn element-wise functions into functions operating on trcks.AwaitableResultTuple values. The success track functions use the plural map_successes (instead of map_success) because they operate on each element individually. Processing short-circuits on the first trcks.Failure, just as in the synchronous case above.

Example

>>> from trcks import AwaitableResultTuple, AwaitableSuccessTuple
>>> from trcks.fp.monads import awaitable_result_tuple as art
>>>
>>> ReadErrorLiteral = Literal["read error"]
>>> WriteErrorLiteral = Literal["write error"]
>>> async def read_from_disk(path: str) -> Result[ReadErrorLiteral, str]:
...     if path != "a.txt" and path != "b.txt":
...         return "failure", "read error"
...     await asyncio.sleep(0.001)
...     contents = {"a.txt": "Hello", "b.txt": "World"}
...     return "success", contents[path]
...
>>> def transform(s: str) -> str:
...     return f"Length: {len(s)}"
...
>>> async def write_to_disk(
...     s: str, path: str
... ) -> Result[WriteErrorLiteral, None]:
...     if path != "output.txt":
...         return "failure", "write error"
...     await asyncio.sleep(0.001)
...     print(f"Wrote '{s}' to file {path}.")
...     return "success", None
...
>>> async def read_and_transform_and_write(
...     input_paths: tuple[str, ...], output_path: str
... ) -> ResultTuple[ReadErrorLiteral | WriteErrorLiteral, str]:
...     p: Pipeline4[
...         tuple[str, ...],
...         AwaitableSuccessTuple[str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral | WriteErrorLiteral, str],
...     ] = (
...         input_paths,
...         art.construct_successes_from_iterable,
...         art.map_successes_to_awaitable_result(read_from_disk),
...         art.map_successes(transform),
...         art.tap_successes_to_awaitable_result(
...             lambda s: write_to_disk(s, output_path)
...         ),
...     )
...     return await pipe(p)
...
>>> asyncio.run(
...     read_and_transform_and_write(("a.txt", "b.txt"), "output.txt")
... )
Wrote 'Length: 5' to file output.txt.
Wrote 'Length: 5' to file output.txt.
('success', ('Length: 5', 'Length: 5'))

To understand what is going on here, let us have a look at the individual steps of the chain:

Example

>>> p1: Pipeline1[
...     tuple[str, ...],
...     AwaitableSuccessTuple[str],
... ] = (
...     ("a.txt", "b.txt"),
...     art.construct_successes_from_iterable,
... )
>>> asyncio.run(art.to_coroutine_result_tuple(pipe(p1)))
('success', ('a.txt', 'b.txt'))
>>>
>>> p2: Pipeline2[
...     tuple[str, ...],
...     AwaitableSuccessTuple[str],
...     AwaitableResultTuple[ReadErrorLiteral, str],
... ] = (
...     ("a.txt", "b.txt"),
...     art.construct_successes_from_iterable,
...     art.map_successes_to_awaitable_result(read_from_disk),
... )
>>> asyncio.run(art.to_coroutine_result_tuple(pipe(p2)))
('success', ('Hello', 'World'))
>>>
>>> p3: Pipeline3[
...     tuple[str, ...],
...     AwaitableSuccessTuple[str],
...     AwaitableResultTuple[ReadErrorLiteral, str],
...     AwaitableResultTuple[ReadErrorLiteral, str],
... ] = (
...     ("a.txt", "b.txt"),
...     art.construct_successes_from_iterable,
...     art.map_successes_to_awaitable_result(read_from_disk),
...     art.map_successes(transform),
... )
>>> asyncio.run(art.to_coroutine_result_tuple(pipe(p3)))
('success', ('Length: 5', 'Length: 5'))
>>>
>>> p4: Pipeline4[
...     tuple[str, ...],
...     AwaitableSuccessTuple[str],
...     AwaitableResultTuple[ReadErrorLiteral, str],
...     AwaitableResultTuple[ReadErrorLiteral, str],
...     AwaitableResultTuple[ReadErrorLiteral | WriteErrorLiteral, str],
... ] = (
...     ("a.txt", "b.txt"),
...     art.construct_successes_from_iterable,
...     art.map_successes_to_awaitable_result(read_from_disk),
...     art.map_successes(transform),
...     art.tap_successes_to_awaitable_result(
...         lambda s: write_to_disk(s, "output.txt")
...     ),
... )
>>> asyncio.run(art.to_coroutine_result_tuple(pipe(p4)))
Wrote 'Length: 5' to file output.txt.
Wrote 'Length: 5' to file output.txt.
('success', ('Length: 5', 'Length: 5'))

Note

The function trcks.fp.monads.awaitable_result_tuple.construct_successes_from_iterable converts a collections.abc.Iterable into a tuple and wraps it into a trcks.AwaitableSuccessTuple, which can then be used with the higher-order functions from trcks.fp.monads.awaitable_result_tuple.

The values pipe(p1), pipe(p2), pipe(p3), and pipe(p4) are all of type trcks.AwaitableResultTuple. Since asyncio.run expects the input type collections.abc.Coroutine, we use the function trcks.fp.monads.awaitable_result_tuple.to_coroutine_result_tuple to convert the trcks.AwaitableResultTuples to collections.abc.Coroutines.

The higher-order functions trcks.fp.monads.awaitable_result_tuple.tap_failure and trcks.fp.monads.awaitable_result_tuple.tap_successes allow us to execute synchronous side effects in the failure case or in the success case (for each element), respectively:

Example

>>> from trcks.fp.composition import Pipeline7
>>> async def read_from_disk(path: str) -> Result[ReadErrorLiteral, str]:
...     if path != "a.txt" and path != "b.txt":
...         return "failure", "read error"
...     await asyncio.sleep(0.001)
...     contents = {"a.txt": "Hello", "b.txt": "World"}
...     return "success", contents[path]
...
>>> async def write_to_disk(
...     s: str, path: str
... ) -> Result[WriteErrorLiteral, None]:
...     if path != "output.txt":
...         return "failure", "write error"
...     await asyncio.sleep(0.001)
...     return "success", None
...
>>> async def read_and_transform_and_write(
...     input_paths: tuple[str, ...], output_path: str
... ) -> ResultTuple[ReadErrorLiteral | WriteErrorLiteral, str]:
...     pipeline: Pipeline7[
...         tuple[str, ...],
...         AwaitableSuccessTuple[str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral | WriteErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral | WriteErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral | WriteErrorLiteral, str],
...     ] = (
...         input_paths,
...         art.construct_successes_from_iterable,
...         art.map_successes_to_awaitable_result(read_from_disk),
...         art.tap_successes(lambda s: print(f"LOG: Read '{s}' from disk.")),
...         art.map_successes(transform),
...         art.tap_successes_to_awaitable_result(
...             lambda s: write_to_disk(s, output_path)
...         ),
...         art.tap_successes(lambda _: print("LOG: Successfully wrote to disk.")),
...         art.tap_failure(lambda err: print(f"LOG: Failed with error: {err}")),
...     )
...     return await pipe(pipeline)
...
>>> result_1 = asyncio.run(
...     read_and_transform_and_write(("a.txt", "b.txt"), "output.txt")
... )
LOG: Read 'Hello' from disk.
LOG: Read 'World' from disk.
LOG: Successfully wrote to disk.
LOG: Successfully wrote to disk.
>>> result_1
('success', ('Length: 5', 'Length: 5'))
>>>
>>> result_2 = asyncio.run(
...     read_and_transform_and_write(("missing.txt",), "output.txt")
... )
LOG: Failed with error: read error
>>> result_2
('failure', 'read error')

Sometimes, side effects themselves can fail and need to return a trcks.AwaitableResult type. The higher-order function trcks.fp.monads.awaitable_result_tuple.tap_successes_to_awaitable_result allows us to execute such asynchronous side effects for each element in the success case. If the side effect returns a trcks.Failure for any element, that failure is propagated. If the side effect returns a trcks.Success for all elements, the original success values are preserved:

Example

>>> async def write_to_disk(s: str) -> Result[OutOfDiskSpace, None]:
...     await asyncio.sleep(0.001)
...     if len(s) > 10:
...         return "failure", "Out of disk space"
...     return "success", None
...
>>> async def read_from_disk(path: str) -> Result[ReadErrorLiteral, str]:
...     if path != "a.txt" and path != "b.txt":
...         return "failure", "read error"
...     await asyncio.sleep(0.001)
...     contents = {"a.txt": "Hi", "b.txt": "Hello, world!"}
...     return "success", contents[path]
...
>>> async def read_and_persist(
...     input_paths: tuple[str, ...],
... ) -> ResultTuple[ReadErrorLiteral | OutOfDiskSpace, str]:
...     pipeline: Pipeline4[
...         tuple[str, ...],
...         AwaitableSuccessTuple[str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral, str],
...         AwaitableResultTuple[ReadErrorLiteral | OutOfDiskSpace, str],
...     ] = (
...         input_paths,
...         art.construct_successes_from_iterable,
...         art.map_successes_to_awaitable_result(read_from_disk),
...         art.tap_successes(lambda s: print(f"LOG: Persisting '{s}'.")),
...         art.tap_successes_to_awaitable_result(write_to_disk),
...     )
...     return await pipe(pipeline)
...
>>> result = asyncio.run(read_and_persist(("a.txt", "b.txt")))
LOG: Persisting 'Hi'.
LOG: Persisting 'Hello, world!'.
>>> result
('failure', 'Out of disk space')