A way to join the results from multiple stages to a single function fusing the results into the positional args of the target function.
Thinking something like this could work, let me know if you think it's worth cleaning up/documenting and writing all the tests for:
from typing import Iterable, Callable
import inspect
from returns.result import Success, Failure, Result
def fuse(results: Iterable[Result], target: Callable) -> Result:
"""Fuse the results from multiple containers into the args for a target function.
Args:
results (Iterable[Result]): The results from multiple container functions.
target (Callable): The target function that receives the arguments.
"""
successes, failures, invalid = [], [], []
for res in results:
match res:
case Success(value):
successes.append(value)
case Failure(error):
failures.append(error)
case _:
invalid.append(res)
if invalid:
return Failure(ValueError(f"Input args are not all Result types: {invalid}"))
if failures:
return Failure(ValueError(f"Not all results are Success: {failures}"))
expected_args = len(inspect.signature(target).parameters)
if len(successes) != expected_args:
return Failure(
ValueError(
f"Expected {expected_args} args for `{target.__name__}`, but got {len(successes)}"
)
)
try:
return target(*successes)
except Exception as e:
return Failure(e)Using this basic case function here:
def add(a, b) -> Result[int, str]:
if a > 0:
return Success(a + b)
return Failure(f"Err: expected `a` > 0. Got `a`= {a}.")Usage:
Happy path all args are expected
fuse([Success(1), Success(3)], add)
output:
Returns failure if some exist
fuse([Success(3), Failure("Err: expected `a` > 0. Got `a`= -2.")], add)output:
<Failure: Not all results are Success: ['Err: expected `a` > 0. Got `a`= -2.']>
Catch all the failures and report them all (avoiding solving one issue then finding another)
fuse([Failure("Err: expected `a` > 0. Got `a`= -2."), Failure("Err: expected `a` > 0. Got `a`= -2.")], add)output:
<Failure: Not all results are Success: ['Err: expected `a` > 0. Got `a`= -2.', 'Err: expected `a` > 0. Got `a`= -2.']>
Catch when someone is passing in a non Result type:
fuse([1, Failure("Err: expected `a` > 0. Got `a`= -2.")], add)output:
<Failure: Input args are not all Result types: [1]>
catch if not given the right number of args:
output:
<Failure: Expected 2 args for `add`, but got 1>
catch overall exceptions just in case:
fuse([Success(1), Success("1")], add)output
<Failure: unsupported operand type(s) for +: 'int' and 'str'>
A way to join the results from multiple stages to a single function fusing the results into the positional args of the target function.
Thinking something like this could work, let me know if you think it's worth cleaning up/documenting and writing all the tests for:
Using this basic case function here:
Usage:
Happy path all args are expected
output:
Returns failure if some exist
output:
Catch all the failures and report them all (avoiding solving one issue then finding another)
output:
Catch when someone is passing in a non Result type:
output:
catch if not given the right number of args:
output:
catch overall exceptions just in case:
output