Playing with Python's operator overloading to support the pipe functional operator from Elixir.

script.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
""" Small experiment with Python's operator overloading to add support for
composing function with the pipe operator similar to Elixir's pipe operator.

The main use case of such technique I see is to build isolated DSL.

This approach uses decorators to achieve the given result which forces you
to wrap all the functions you want to use this way. The runtime cost should
be minimal however this might be a problem in some context. It also prevents you
from using arbitrary functions.

Another approach I found after making this can be found at
https://hackernoon.com/adding-a-pipe-operator-to-python-19a3aa295642 and only
requires wrapping the callsite. """

import functools as ft
import inspect

def _chain(*callables):
    return ft.partial(ft.reduce, lambda acc, func: func(acc), callables)


class PipingCallable(object):

    __slots__ = ("_callable",)

    def __init__(self, func):
        # For this simple experiment, functions with multiple arguments
        # and / or keywords arguments need to be wrapped externally using
        # partial. We could combine this with currying to get a similar syntax
        # as the one used in Elixir.
        assert len(inspect.signature(func).parameters) == 1, ''
        self._callable = func

    def __call__(self, *args, **kwargs):
        return self._callable(*args, **kwargs)

    def __ror__(self, lhe):
        if callable(lhe):
            return partial(_chain(self._callable, lhe))
        return self(lhe)

    def __or__(self, rhe):
        assert callable(rhe)
        return partial(_chain(rhe, self._callable))


# This function implements the same behaviour as `PipingCallable` without the
# operator overloading trick.
def pipe(value, *callables):
    return _chain(*callables)(value)


# Use this to turn functions with multiple arguments into functions suitable for
# use in the pipe.
def partial(func, *args, **kwargs):
    return PipingCallable(ft.partial(func, *args, **kwargs))


# Decorator shortcurt
composable = partial


@composable
def squares(arr):
    return [x * x for x in arr]


@composable
def odds(arr):
    return [x for x in arr if x % 2 != 0]


def add(value, arr):
    return [x + value for x in arr]


if __name__ == "__main__":
    # Using the pipe operator composition
    assert (
        [1, 2, 3, 4]
        | squares
        | odds
        | partial(add, 1)
    ) == [2, 10]

    # Without operator overloading
    assert pipe([1, 2, 3, 4], squares, odds, partial(add, 1)) == [2, 10]

    # Without anything
    assert add(1, odds(squares([1, 2, 3, 4]))) == [2, 10]