feat request/suggestion: fuse results from multiple results into the args of one target function

Bắt đầu bằng cách xem xét các API Result, Success và Failure hiện có cùng các bài kiểm thử hiện tại của chúng; issue không nêu tên tệp hoặc điểm vào. So sánh hành vi fuse được đề xuất với các quy ước về kiểu và xử lý lỗi của dự án. Được xem là hoàn tất khi một API đã thống nhất được tích hợp, lập tài liệu và được bao phủ bởi các bài kiểm thử cho trường hợp thành công, lỗi, đầu vào không hợp lệ, số lượng đối số và các ngoại lệ của target.

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:

<Success: 4>

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:

fuse([Success(1)], add)

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'>