Improve the ergonomics of pipelines

[mardukbp]

Is your feature request related to a problem? Please describe. In F# one can write the following pipeline:

let result =
  (val1, val2, val3)
  |> convertVal1  // A * B * C -> (P * Q) * B * C
  |> convertVal2  // (P * Q) * B * C -> S * C
  |> convertVal3  // S * C -> S * T

With Expression I was able to implement it in two non-ergonomic ways:

result = pipe(
    (val1, val2, val3),
    lambda x: starpipe(x, convertVal1),
    lambda x: starpipe(x, convertVal2),
    lambda x: starpipe(x, convertVal3)
)

result = starpipe(
    starpipe(
        starpipe(
            (val1, val2, val3),
            convertVal1
        ),
        convertVal2
    ),
    convertVal3
)

Describe the solution you'd like I would like to write the same code as in F#.

Describe alternatives you've considered Perhaps I am missing something and Expression does allow writing something similar to the F# code.

Additional context

[mardukbp]

Never mind. I figured out a simple solution:

result = pipe(
  convertVal1(val1, val2, val3),
  convertVal2,
  convertVal3
)

Thank you for writing this awesome library!

[mardukbp]

Spoke too soon. It typechecks but it does not work at runtime.

What works is:

result = pipe(
    (val1, val2, val3),
    lambda t: convertVal1(*t),
    lambda t: convertVal2(*t),
    lambda t: convertVal3(*t)
)

I think it would be good to have a function multipipe for piping functions that have multiple return values.

In F# the above pipe works because the type signature of the functions is A * B -> C * D. But in Python the type signature is Callable[[A,B], tuple[C, D]], which is not composable.

[dbrattli]

Could you please post a minimum and complete example?

[mardukbp]

Sure. Sorry for the slopiness.

from expression import pipe

def f(a, b):
    return a, b

# does not typecheck, but in F# it would
# result = pipe(
#   (1, 2),
#    f,
#    f
# )

# workaround that could be simplified with multipipe
result = pipe(
    (1, 2),
    lambda t: f(*t),
    lambda t: f(*t),
)

[dbrattli]

If F# the function f would take a tuple as input. That works in Python as well:

def f(t: tuple[_A, _B]) -> tuple[_A, _B]:
    return t

result = pipe(
  (1, 2),
   f,
   f
)

But I also think this should work:

def f(a: _A, b: _B) -> tuple[_A, _B]:
    return a, b

result = starpipe(
    (1, 2),
    f,
    f,
)

Let me check ...

[dbrattli]

The #194 PR should make it possible to starpipe through tuple generating functions that take N arguments. Would it be possible for you to test this locally to see if it fixes your problems? E.g the current tests:

def fn(a: _A, b: _B) -> tuple[_A, _B]:
    return a, b

def gn(a: _A, b: _B) -> tuple[_B, _A]:
    return b, a

def yn(a: _A, b: _B) -> tuple[_A, _B, int]:
    return a, b, 3

def test_starpipe_simple():
    assert starpipe((1, 2), fn) == fn(1, 2)

def test_starpipe_id():
    assert starpipe((1, 2), starid) == (1, 2)

def test_starpipe_fn_gn():
    assert starpipe((1, 2), fn, gn) == gn(*fn(1, 2))

def test_starpipe_fn_gn_yn():
    assert starpipe((1, 2), fn, gn, yn) == yn(*gn(*fn(1, 2)))

[mardukbp]

Thank you for your efforts. Unfortunately your solution does not typecheck. The reason is that starpipe expects functions with a variable number of arguments. But tuples are not lists. They are products of exactly N types.

I did some experimentation and came up with a solution that is ergonomic and works for arbitrarily long pipelines:

from typing import Callable, Iterable, TypeVar
from functools import reduce

T = TypeVar("T")

def compose(f: Callable[..., tuple[T, ...]], g: Callable[..., tuple[T, ...]]):
    def h(*xs) -> tuple[T, ...]:
        return g(*f(*xs))
    return h

def pipe3(xs: tuple[T, T, T], functions: Iterable[Callable[[T, T, T], tuple[T, T, T]]]):
    f: Callable[[T, T, T], tuple[T, T, T]] = reduce(compose, functions)
    return f(*xs)

def f(a: int, b: int, c: int):
    return a, b, c

def g(a: int, b: int, c: int):
    return a*a, b*b, c*c

result = pipe3(
    (1, 2, 3),
    (
        f,
        g,
        f,
        g
    )
)

Inspired by C# one could define pipe1 through pipe15 and that should cover most cases. What do you think?

[dbrattli]

Sorry, I don't get it. Starpipe do not expect a function with variable number of arguments. It just tells that if you give an n-tuple then then function must take n arguments, and if the function produces an m-tuple, then the next function must take m arguments. Can you show me the failing example?

The example above type-checks just fine:

from expression.core.pipe import starpipe

def f(a: int, b: int, c: int) -> tuple[int, int, int]:
    return a, b, c

def g(a: int, b: int, c: int) -> tuple[int, int, int]:
    return a * a, b * b, c * c

result = starpipe(
    (1, 2, 3),
    f,
    g,
    f,
    g,
)

print("result:", result)

PS: Did you try the latest version with the new starpipe?

[mardukbp]

I installed the commit 3cfd9e8 and this is what MyPy reports:

I think this comes from this overload:

@overload
def starpipe(
    __args: tuple[*_P],
    __fn1: Callable[[*_P], tuple[*_Q]],
    __fn2: Callable[[*_Q], tuple[*_X]],
    __fn3: Callable[[*_X], _B],
) -> _B:
    ...

Callable[[*_P], tuple[*_Q]] is a function that expectes a variable number of arguments and returns a tuple.

Starpipe do not expect a function with variable number of arguments. It just tells that if you give an n-tuple then then function must take n arguments, and if the function produces an m-tuple, then the next function must take m arguments.

This is definitely not what the above type signature means.

[dbrattli]

I don't get any errors with mypy-1.8.0. What version of mypy are you using?

This is definitely not what the above type signature means.

Ok, can you please explain when you have such a strong opinion about this? IMO the type signature says that the type is variadic not the function.

[mardukbp]

You are right. MyPy 1.8.0 does not complain. The latest release of the MyPy VS Code extension bundles MyPy 1.6.1. Thank you for your help :)

Ok, can you please explain when you have such a strong opinion about this?

@overload
def starpipe(
    __args: tuple[*_P],
    __fn1: Callable[[*_P], tuple[*_Q]],
    __fn2: Callable[[*_Q], tuple[*_X]],
    __fn3: Callable[[*_X], _B],
) -> _B:
    ...

Nowhere in the type signatures is specified that a function takes or returns a specific number of values. The star just captures everything. The following error message (from MyPy 1.8.0) does not really help understand what is wrong with the pipeline:

It does not say that a function of two arguments was expected.

In any case what matters is that the issue has been solved. Thank you for your time and dedication :)

[dbrattli]

PS: Pylance/pyright has better error message than mypy:

Argument of type "(a: int, b: int, c: int) -> tuple[int, int, int]" cannot be assigned to parameter "__fn2" of type "(*_Q@starpipe) -> _B@starpipe" in function "starpipe"
  Type "(a: int, b: int, c: int) -> tuple[int, int, int]" cannot be assigned to type "(int, int) -> tuple[int, int, int]"
    Function accepts too few positional parameters; expected 3 but received 2Pylance[reportGeneralTypeIssues](https://github.com/microsoft/pyright/blob/main/docs/configuration.md#reportGeneralTypeIssues)

[mardukbp]

Awesome! Thank you!