Tuple code with trcks.oop
See also
The functional tuple page covers the same operations using function composition.
Synchronous single-track code with trcks.oop.TupleWrapper
While the class trcks.oop.Wrapper wraps and operates on a single value, the class trcks.oop.TupleWrapper wraps a homogeneous tuple and applies operations to each element individually. This is useful when we need to process multiple values through the same chain of transformations:
Example
>>> from trcks.oop import TupleWrapper
>>>
>>> def normalize_email(email: str) -> str:
... return email.strip().lower()
>>>
>>> def to_domain(email: str) -> str:
... return email.split("@")[1]
>>>
>>> emails = (" Erika.Mustermann@Domain.ORG ", "JOHN_DOE@Provider.COM ")
>>> (TupleWrapper.construct_from_iterable(emails).map(normalize_email).map(to_domain).core)
('domain.org', 'provider.com')
To understand what is going on here, let us have a look at the individual steps of the chain:
Step by step
>>> # 1. Wrap the input tuple:
>>> wrapped: TupleWrapper[str] = TupleWrapper.construct_from_iterable(
... (" Erika.Mustermann@Domain.ORG ", "JOHN_DOE@Provider.COM ")
... )
>>> wrapped
TupleWrapper(core=(' Erika.Mustermann@Domain.ORG ', 'JOHN_DOE@Provider.COM '))
>>> # 2. Apply normalize_email to each element:
>>> mapped: TupleWrapper[str] = wrapped.map(normalize_email)
>>> mapped
TupleWrapper(core=('erika.mustermann@domain.org', 'john_doe@provider.com'))
>>> # 3. Apply to_domain to each element:
>>> mapped_again: TupleWrapper[str] = mapped.map(to_domain)
>>> mapped_again
TupleWrapper(core=('domain.org', 'provider.com'))
>>> # 4. Unwrap the output tuple:
>>> unwrapped: tuple[str, ...] = mapped_again.core
>>> unwrapped
('domain.org', 'provider.com')
Note
trcks.oop.TupleWrapper.construct wraps a single value
in a one-element tuple:
TupleWrapper.construct(42) produces TupleWrapper(core=(42,)).
The method trcks.oop.TupleWrapper.map_to_iterable applies a function that returns a homogeneous tuple to each element and flattens the results (like a "flat map"):
>>> TupleWrapper.construct_from_iterable(("ab", "cd")).map_to_iterable(tuple).core
('a', 'b', 'c', 'd')
The tap method allows executing side effects for each element
while preserving the original tuple:
Example
>>> tuple_wrapper = (
... TupleWrapper.construct_from_iterable(emails)
... .map(normalize_email)
... .tap(lambda e: print(f"LOG: Processing '{e}'."))
... .map(to_domain)
... )
LOG: Processing 'erika.mustermann@domain.org'.
LOG: Processing 'john_doe@provider.com'.
>>> tuple_wrapper.core
('domain.org', 'provider.com')
Synchronous double-track code with trcks.oop.ResultTupleWrapper
When applying a failable function to each element in a tuple,
we need the trcks.oop.ResultTupleWrapper class.
The success track methods are named map_successes and tap_successes (plural)
because they operate on each element in the trcks.SuccessTuple individually.
Processing short-circuits on the first trcks.Failure.
Example
>>> from typing import Literal
>>> from trcks import Result, ResultTuple
>>> from trcks.oop import TupleWrapper
>>>
>>> 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]:
... return (
... TupleWrapper.construct_from_iterable(user_emails)
... .map_to_result(get_user_id)
... .map_successes_to_result(get_subscription_id)
... .map_successes(get_subscription_fee)
... .core
... )
>>>
>>> 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:
Step by step
>>> from trcks.oop import ResultTupleWrapper
>>>
>>> # 1. Wrap the input tuple:
>>> wrapped: TupleWrapper[str] = TupleWrapper.construct_from_iterable(
... ("erika.mustermann@domain.org",)
... )
>>> wrapped
TupleWrapper(core=('erika.mustermann@domain.org',))
>>> # 2. Apply the Result function get_user_id to each element:
>>> mapped_once: ResultTupleWrapper[UserDoesNotExist, int] = wrapped.map_to_result(
... get_user_id
... )
>>> mapped_once
ResultTupleWrapper(core=('success', (1,)))
>>> # 3. Apply the Result function get_subscription_id to each element:
>>> mapped_twice: ResultTupleWrapper[FailureDescription, int] = (
... mapped_once.map_successes_to_result(get_subscription_id)
... )
>>> mapped_twice
ResultTupleWrapper(core=('success', (42,)))
>>> # 4. Apply get_subscription_fee to each element:
>>> mapped_thrice: ResultTupleWrapper[FailureDescription, float] = (
... mapped_twice.map_successes(get_subscription_fee)
... )
>>> mapped_thrice
ResultTupleWrapper(core=('success', (4.2,)))
>>> # 5. Unwrap the output result tuple:
>>> unwrapped: ResultTuple[FailureDescription, float] = mapped_thrice.core
>>> unwrapped
('success', (4.2,))
Note
The method trcks.oop.TupleWrapper.map_to_result
returns a trcks.oop.ResultTupleWrapper object.
The corresponding class trcks.oop.ResultTupleWrapper
has a map_failure* and a map_successes* method
for each map* method of the class trcks.oop.TupleWrapper.
Note the plural map_successes (instead of map_success)
since the method operates on each element in the tuple.
The tap_successes and tap_failure methods allow us to execute side effects
in the success case (for each element) or in the failure case, respectively:
Example
>>> def get_subscription_fees_by_email(
... user_emails: tuple[str, ...],
... ) -> ResultTuple[FailureDescription, float]:
... return (
... TupleWrapper.construct_from_iterable(user_emails)
... .map_to_result(get_user_id)
... .tap_successes(lambda n: print(f"LOG: User ID: {n}."))
... .map_successes_to_result(get_subscription_id)
... .map_successes(get_subscription_fee)
... .tap_successes(lambda x: print(f"LOG: Subscription fee: {x}."))
... .tap_failure(lambda fd: print(f"LOG: Failure: {fd}."))
... .core
... )
>>>
>>> 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 tap_successes_to_result method 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]:
... return (
... TupleWrapper.construct_from_iterable(user_emails)
... .map_to_result(get_user_id)
... .tap_successes_to_result(write_to_disk)
... .core
... )
>>>
>>> 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)
>>>
>>> (
... TupleWrapper.construct_from_iterable(
... (
... "erika.mustermann@domain.org",
... "jane_doe@provider.com",
... "john_doe@provider.com",
... )
... )
... .map_to_result(get_user_id_logged)
... .core
... )
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 with trcks.oop.AwaitableTupleWrapper
While the class trcks.oop.TupleWrapper and its method map
allow the chaining of synchronous functions for each element,
they cannot chain asynchronous functions.
The method trcks.oop.TupleWrapper.map_to_awaitable
and the class trcks.oop.AwaitableTupleWrapper
allow us to combine collections.abc.Awaitable-returning functions
with other collections.abc.Awaitable-returning functions or
with "regular" functions,
applied to each element in the tuple:
Example
>>> import asyncio
>>>
>>> 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, ...]:
... return await (
... TupleWrapper.construct_from_iterable(input_paths)
... .map_to_awaitable(read_from_disk)
... .map(transform)
... .core
... )
>>>
>>> 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:
Step by step
>>> from trcks.oop import AwaitableTupleWrapper
>>> # 1. Wrap the input tuple:
>>> wrapped: TupleWrapper[str] = TupleWrapper.construct_from_iterable(("a.txt", "b.txt"))
>>> wrapped
TupleWrapper(core=('a.txt', 'b.txt'))
>>> # 2. Apply the asynchronous function read_from_disk to each element:
>>> mapped_once: AwaitableTupleWrapper[str] = wrapped.map_to_awaitable(read_from_disk)
>>> mapped_once
AwaitableTupleWrapper(core=<coroutine object ...>)
>>> # 3. Apply the function transform to each element:
>>> mapped_twice: AwaitableTupleWrapper[str] = mapped_once.map(transform)
>>> mapped_twice
AwaitableTupleWrapper(core=<coroutine object ...>)
>>> # 4. Unwrap and run the output coroutine:
>>> asyncio.run(mapped_twice.core_as_coroutine)
('Length: 5', 'Length: 5')
Note
The property core of the class trcks.oop.AwaitableTupleWrapper
has type trcks.AwaitableTuple.
Since asyncio.run expects a collections.abc.Coroutine object,
we need to use the property core_as_coroutine instead.
The method trcks.oop.AwaitableTupleWrapper.tap allows us to execute synchronous side effects for each element. Similarly, the method trcks.oop.AwaitableTupleWrapper.tap_to_awaitable allows us to execute asynchronous side effects for each element.
Example
>>> 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, ...]:
... return await (
... TupleWrapper.construct_from_iterable(input_paths)
... .map_to_awaitable(read_from_disk)
... .tap(lambda s: print(f"Read '{s}' from disk."))
... .map(transform)
... .tap(lambda s: print(f"Transformed to '{s}'."))
... .core
... )
>>>
>>> result = 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'.
>>> result
('Length: 5', 'Length: 5')
The widening methods
trcks.oop.AwaitableTupleWrapper.map_to_result,
trcks.oop.AwaitableTupleWrapper.map_to_result_iterable,
trcks.oop.AwaitableTupleWrapper.map_to_awaitable_result, and
trcks.oop.AwaitableTupleWrapper.map_to_awaitable_result_iterable
of the class trcks.oop.AwaitableTupleWrapper
apply failable functions to each element individually
and widen the wrapped trcks.AwaitableTuple into
a trcks.AwaitableResultTuple
(see the glossary).
Similarly, the methods tap_to_result, tap_to_result_iterable,
tap_to_awaitable_result, and tap_to_awaitable_result_iterable
widen the wrapper based on the outcome of failable side effects.
Processing short-circuits on the first trcks.Failure:
Example
>>> async def get_subscription_fees_slowly(
... user_emails: tuple[str, ...],
... ) -> ResultTuple[FailureDescription, float]:
... return await (
... AwaitableTupleWrapper.construct_from_iterable(user_emails)
... .map_to_result(get_user_id)
... .map_successes_to_result(get_subscription_id)
... .map_successes(get_subscription_fee)
... .core
... )
>>>
>>> asyncio.run(get_subscription_fees_slowly(("erika.mustermann@domain.org",)))
('success', (4.2,))
>>> asyncio.run(get_subscription_fees_slowly(("john_doe@provider.com",)))
('failure', 'User does not have a subscription')
>>> asyncio.run(get_subscription_fees_slowly(("jane_doe@provider.com",)))
('failure', 'User does not exist')
Asynchronous double-track code with trcks.oop.AwaitableResultTupleWrapper
Whenever we define a function using
the async def ... -> Result[F, S] syntax
and want to apply it to each element in a tuple,
we need the trcks.oop.AwaitableResultTupleWrapper class.
The package trcks provides the type alias
trcks.AwaitableResultTuple[F, S]
for Awaitable[ResultTuple[F, S]].
The success track methods are named map_successes and tap_successes (plural)
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
>>>
>>> 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]:
... return await (
... TupleWrapper.construct_from_iterable(input_paths)
... .map_to_awaitable_result(read_from_disk)
... .map_successes(transform)
... .tap_successes_to_awaitable_result(lambda s: write_to_disk(s, output_path))
... .core
... )
>>>
>>> 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:
Step by step
>>> from trcks.oop import AwaitableResultTupleWrapper
>>> # 1. Wrap the input tuple:
>>> wrapped: TupleWrapper[str] = TupleWrapper.construct_from_iterable(("a.txt", "b.txt"))
>>> wrapped
TupleWrapper(core=('a.txt', 'b.txt'))
>>> # 2. Apply the AwaitableResult function read_from_disk to each element:
>>> mapped_once: AwaitableResultTupleWrapper[ReadErrorLiteral, str] = (
... wrapped.map_to_awaitable_result(read_from_disk)
... )
>>> mapped_once
AwaitableResultTupleWrapper(core=<coroutine object ...>)
>>> # 3. Apply the function transform to each element in the success case:
>>> mapped_twice: AwaitableResultTupleWrapper[ReadErrorLiteral, str] = (
... mapped_once.map_successes(transform)
... )
>>> mapped_twice
AwaitableResultTupleWrapper(core=<coroutine object ...>)
>>> # 4. Apply the failable async side effect write_to_disk to each element:
>>> mapped_thrice: AwaitableResultTupleWrapper[
... ReadErrorLiteral | WriteErrorLiteral, str
... ] = mapped_twice.tap_successes_to_awaitable_result(
... lambda s: write_to_disk(s, "output.txt")
... )
>>> mapped_thrice
AwaitableResultTupleWrapper(core=<coroutine object ...>)
>>> # 5. Unwrap and run the output coroutine:
>>> asyncio.run(mapped_thrice.core_as_coroutine)
Wrote 'Length: 5' to file output.txt.
Wrote 'Length: 5' to file output.txt.
('success', ('Length: 5', 'Length: 5'))
The methods trcks.oop.AwaitableResultTupleWrapper.tap_failure and trcks.oop.AwaitableResultTupleWrapper.tap_successes allow us to execute synchronous side effects in the failure case or in the success case (for each element), respectively:
Example
>>> 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]:
... return await (
... TupleWrapper.construct_from_iterable(input_paths)
... .map_to_awaitable_result(read_from_disk)
... .tap_successes(lambda s: print(f"LOG: Read '{s}' from disk."))
... .map_successes(transform)
... .tap_successes_to_awaitable_result(lambda s: write_to_disk(s, output_path))
... .tap_successes(lambda _: print("LOG: Successfully wrote to disk."))
... .tap_failure(lambda err: print(f"LOG: Failed with error: {err}"))
... .core
... )
>>>
>>> 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 method trcks.oop.AwaitableResultTupleWrapper.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]:
... return await (
... TupleWrapper.construct_from_iterable(input_paths)
... .map_to_awaitable_result(read_from_disk)
... .tap_successes(lambda s: print(f"LOG: Persisting '{s}'."))
... .tap_successes_to_awaitable_result(write_to_disk)
... .core
... )
>>>
>>> result = asyncio.run(read_and_persist(("a.txt", "b.txt")))
LOG: Persisting 'Hi'.
LOG: Persisting 'Hello, world!'.
>>> result
('failure', 'Out of disk space')