Python Pipes

MC706.io
Functional Programming, Monkey Patching, Pipes, Python

One of my favorite parts of the Elm programming language is its ability to pipe functions in a nice and clean syntax as so:

1
2
3
4
5
1
    |> add 1
    |> multiply 2
    |> div 4
    |> multiply 10

I covered this in my previous post on functionanl programming in python where I build a evaluate_pipeline funciton that had a similar function:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from functools import reduce

def compose(*functions):
    return reduce(lambda f, g: lambda x: f(g(x)), functions, lambda x: x)

def pipeline(*functions):
    return compose(*reversed(functions))

def evaluate_pipeline(start, *functions):
    return pipeline(*functions)(start)

evaluate_pipeline(
    1,
    partial(add, 1),
    partial(multiply, 2),
    partial(div, 4)
    partial(multiply, 10),
)

Recently I have been feeling the lack of python’s functional capabilities so I went looking for a cleaner pipeline in python.

What I found was both wonder and scary. It creates a class to hook into pythons underlying “dunder” methods to allow for a pipeline that looks like this:

1
2
3
4
5
6
(1
    >> add(1)
    >> multiply(2)
    >> div(4)
    >> multipy(10)
)

The syntax is pretty, but extremely foreign to python. Lets take a look at how it works.

Pipe Decorator

The Key here is all of the functions that exist in the pipeline have to be decorated. The decorator wraps the functions in a class that implements __rrshift__ allowing us to define custom handling of the bitwise shift right operator.

1
2
3
4
5
6
7
8
9
10
11
12
13
def pipe(func: Callable):
    class Pipe:
        def __init__(self, *args, **kwargs):
            self.args = args
            self.kwargs = kwargs

        def __rrshift__(self, other):
            return func(other, *self.args, **self.kwargs)

        def __call__(self, other):
            return other >> self

    return Pipe

This decorator then wraps any function with Pipe, which knows how to take a value and apply it and pass along.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@pipe
def add(a: int, b: int) -> int:
    return a + b

@pipe
def multiply(a: int, b: int) -> int:
    return a * b

@pipe
def div(a: int, b: int -> int:
    return a / b

@pipe
def debug(value: Any) -> Any:
    print(value)
    return value

(1
    >> add(1)
    >> multiply(2)
    >> div(4)
    >> muliplty(10)
    >> debug()
)

Caveats

This works great for single arity functions, or with functions that all take a common first argument and return one of a similiar type. In fact, I found this as a hack for pandas.DataFrame, creating pipeable operations that all take a data frame as a first example.

Since each item in the pipeline is now a Pipe object instead of a function, you cannot natively call @pipe decorated functions, however to allow for this interop, they can be applied at call time.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def add(a: int, b: int) -> int:
    return a + b

def multiply(a: int, b: int) -> int:
    return a * b

def div(a: int, b: int -> int:
    return a / b

(1
    >> pipe(add)(1)
    >> pipe(multiply)(2)
    >> pipe(div)(4)
    >> pipe(multiply)(10)
)

however that is not nearly as clean.

OOP Pipeline

We can achieve something similar without the crazy >> syntax:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class Pipline:
    def __init__(self, value):
        self.value = value

    def pipe(self, func, *args, **kwargs):
        return Pipeline(func(self.value, *args, **kwargs))

(Pipeline(1)
    .pipe(add, 1)
    .pipe(multiply, 2)
    .pipe(div, 4)
    .pipe(multiply, 10)
    .value
)

This works basically the same without any crazy python syntax, however you do need to add the .value at the end if you want an actual value, instead of a Pipeline object. This method is also cleaner in that we can take pure python functions as arguments without any need for partial application.

Reduce Pipeline

While we are at it, we can also make a pipeline out of reduce and some s style expression tuples.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from functools import reduce
from typing import List

def pipeline(value, operations: List[tuple]):
    def eval_s_expression(expression_tuple):
        func, *args = expression_tuple
        def evaluate(x):
            return func(x, *args)
        return evaluate
    return reduce(lambda x, y: eval_s_expression(y)(x), operations,value)

pipeline(1, [
    (add, 1),
    (multiply, 2),
    (div, 4),
    (multiply, 10),
])

Very similar to the OOP version in that we dont need partial or any specially decorated functions and here we dont need to call .value at the end to get an actual value. This also has the benefit of being able to dynamically build the operations list of functions and arguments and pass it into the pipeline.

Assumptions

So far all of the pipelines have assumed that every function passed into the pipeline returns a value, and that it takes its first argument as the pipelined value. This is a little different from the elm or haskell way were the last value of a function is the one that is pipelined.

We can fix that in a few places by swapping the pass in order of *args and the value being passed in.

Conclusions

Pipelines are a cool concept and being able to cleanly implement them in a few ways in python was interesting. I am not sure I would ever use the >> pipeline in a production system, but it was an interesting look at using builtins to extend python’s syntax.