Python pipe decorator with Ellipsis

pipe_ellipsis.py

""" A proposal for yet another pipe decorator

I'm still missing a nice way to do method chaining in python.

I'd like to have something like `func_a() | func_b()` as a syntax and I'm quite
aware that there are some "pipe" libraries on pypi like `pipe` or `pipe21` that
offer such functionality.

But what I don't like is, that the result of `func_a` (in the example above) is
piped directly as the first argument to `func_b`.

Let's assume this setup:

    from some_piping_library import pipe

    @pipe
    def func_a(iterable, x):
        ...

    @pipe
    def func_b(iterable, y):
        ...


With the current available libraries, the pipe construct would look like this:

    result = func_a([1, 2, 3], 10) | func_b(20)


And I especially find the term `| func_b(20)` disgusting. The function does
take two parameters, but only one is shown. I know that this behaviour is also
found in other programming languages (looking at you, Ruby) and I really loath
it.

Isn't explicit better than implicit?

It also makes it impossible to use a function with a definition like the
built-in `map` that takes an iterator as a second positional argument or to use
the piped in value as a keyword argument.

I came up with the idea to use the Ellipsis (...) as a marker for the piped
value - it's built into Python and its used to indicate something that has to
be implemented - or in this case evaluated.


So "my" pipe construct would look like this:

    result = func_a([1, 2, 3], 10) | func_b(..., 20)


I think this is much clearer. And it allows to swap places in the arguments:

    result = func_a([1, 2, 3], 10) | map_like(square, ...)


Or use the pipe result as a keyword argument

    result = func_a([1, 2, 3], 10) | func_b(20, iterable=...)


Not quite sure for now, if I should really make this a real project / library.
Maybe I just use it for myself for now and decide later.
    
"""

import functools
import logging

logging.basicConfig(level=logging.DEBUG)


class Pipe:

    def __init__(self, function):
        self._function = function
        self._args = None
        self._kwargs = None
        self._result = None
        functools.update_wrapper(self, function)

    def __call__(self, *args, **kwargs):
        """ the decorated function is called

        If an elipsis (...) is in args or kwargs values, the parameters
        are stored for a later execustion.

        If the elipsis is not present, the result of the function call is
        returned directly.
        """

        logging.debug(f"CALL   {self._function}, {args}, {kwargs}")

        if ... in args or ... in kwargs.values():
            # preparing for pipe
            self._args = args
            self._kwargs = kwargs
            return self
        else:
            # direct execution
            logging.debug(f"DIRECT {self._function}, {args}, {kwargs}")
            return self._function(*args, **kwargs)

    def __ror__(self, left):
        """ execute a prepared decorated function

        The elipsis in args or kargs values is replaced with the actual value
        from the or operation and the decorated function is executed.
        """
        args = tuple(left if v == ... else v for v in self._args)
        kwargs = {k: left if v == ... else v for k, v in self._kwargs.items()}
        logging.debug(f"EXEC   {self._function}, {args}, {kwargs}")
        return self._function(*args, **kwargs)

    def __str__(self):
        """ returns the string representation of the decorated function """
        return str(self._function)

    def __repr__(self):
        """ returns the string representation of the decorated function """
        return repr(self._function)


@Pipe
def add_to(x:int, y:float):
    return x + y

@Pipe
def div_by(x, y):
    return x / y

@Pipe
def power_to(x, pow=2):
    return x ** pow

r = add_to(1, 15) | div_by(..., 2) | power_to(2, pow=...)
print(f"{r=}")

# results in
# DEBUG:root:CALL   <function add_to at 0x10c80df80>, (1, 15), {}
# DEBUG:root:DIRECT <function add_to at 0x10c80df80>, (1, 15), {}
# DEBUG:root:CALL   <function div_by at 0x10c9c5620>, (Ellipsis, 2), {}
# DEBUG:root:EXEC   <function div_by at 0x10c9c5620>, (16, 2), {}
# DEBUG:root:CALL   <function power_to at 0x10c9c56c0>, (2,), {'pow': Ellipsis}
# DEBUG:root:EXEC   <function power_to at 0x10c9c56c0>, (2,), {'pow': 8.0}
# r=256.0

print("")


@Pipe
def iter_add(iterable, y):
    """ test docstring """
    return [item + y for item in iterable]

@Pipe
def iter_div(iterable, by):
    return [item / by for item in iterable]

@Pipe
def iter_power(by, iterable):
    return [item**by for item in iterable]

r = iter_add([2, 4, 6], 2) | iter_div(..., 2) | iter_power(4, iterable=...)
print(f"{sum(r)=}")

# results in
# DEBUG:root:CALL   <function iter_add at 0x10c9c5800>, ([2, 4, 6], 2), {}
# DEBUG:root:DIRECT <function iter_add at 0x10c9c5800>, ([2, 4, 6], 2), {}
# DEBUG:root:CALL   <function iter_div at 0x10c9c58a0>, (Ellipsis, 2), {}
# DEBUG:root:EXEC   <function iter_div at 0x10c9c58a0>, ([4, 6, 8], 2), {}
# DEBUG:root:CALL   <function iter_power at 0x10c9c5940>, (4,), {'iterable': Ellipsis}
# DEBUG:root:EXEC   <function iter_power at 0x10c9c5940>, (4,), {'iterable': [2.0, 3.0, 4.0]}
# sum(r)=353.0