Pipe — Infix programming toolkit

Module enabling a sh like infix syntax (using pipes).

Introduction

As an example, here is the solution for the 2nd Euler Project problem:

Find the sum of all the even-valued terms in Fibonacci which do not exceed four million.

Given fib a generator of Fibonacci numbers:

sum(fib() | where(lambda x: x % 2 == 0) | take_while(lambda x: x < 4000000))

Each pipes is lazy evalatated, can be aliased, and partially initialized, so it could be rewritten as:

is_even = where(lambda x: x % 2 == 0)
sum(fib() | is_even | take_while(lambda x: x < 4000000)

Installing

To install the library, you can just run the following command:

# Linux/macOS
python3 -m pip install pipe

# Windows
py -3 -m pip install pipe

Using

The basic syntax is to use a | like in a shell:

>>> from itertools import count
>>> from pipe import select, take
>>> sum(count() | select(lambda x: x ** 2) | take(10))
285
>>>

Some pipes take an argument:

>>> from pipe import where
>>> sum([1, 2, 3, 4] | where(lambda x: x % 2 == 0))
6
>>>

Some do not need one:

>>> from pipe import traverse
>>> for i in [1, [2, 3], 4] | traverse:
...     print(i)
1
2
3
4
>>>

In which case it's allowed to use the calling parenthesis:

>>> from pipe import traverse
>>> for i in [1, [2, 3], 4] | traverse():
...     print(i)
1
2
3
4
>>>

Existing Pipes in this module

Alphabetical list of available pipes; when several names are listed for a given pipe, these are aliases.

batched

Like Python 3.12 itertool.batched:

>>> from pipe import batched
>>> list("ABCDEFG" | batched(3))
[('A', 'B', 'C'), ('D', 'E', 'F'), ('G',)]
>>>

chain

Chain a sequence of iterables:

>>> from pipe import chain
>>> list([[1, 2], [3, 4], [5]] | chain)
[1, 2, 3, 4, 5]
>>>

Warning : chain only unfolds an iterable containing ONLY iterables:

list([1, 2, [3]] | chain)

Gives a TypeError: 'int' object is not iterable Consider using traverse.

chain_with(other)

Like itertools.chain, yields elements of the given iterable, then yields elements of its parameters

>>> from pipe import chain_with
>>> list((1, 2, 3) | chain_with([4, 5], [6]))
[1, 2, 3, 4, 5, 6]
>>>

dedup(key=None)

Deduplicate values, using the given key function if provided.

>>> from pipe import dedup
>>> list([-1, 0, 0, 0, 1, 2, 3] | dedup)
[-1, 0, 1, 2, 3]
>>> list([-1, 0, 0, 0, 1, 2, 3] | dedup(key=abs))
[-1, 0, 2, 3]
>>>

enumerate(start=0)

The builtin enumerate() as a Pipe:

>>> from pipe import enumerate
>>> list(['apple', 'banana', 'citron'] | enumerate)
[(0, 'apple'), (1, 'banana'), (2, 'citron')]
>>> list(['car', 'truck', 'motorcycle', 'bus', 'train'] | enumerate(start=6))
[(6, 'car'), (7, 'truck'), (8, 'motorcycle'), (9, 'bus'), (10, 'train')]
>>>

filter(predicate)

Alias for where(predicate), see where(predicate).

groupby(key=None)

Like itertools.groupby(sorted(iterable, key = keyfunc), keyfunc)

>>> from pipe import groupby, map
>>> items = range(10)
>>> ' / '.join(items | groupby(lambda x: "Odd" if x % 2 else "Even")
...                  | select(lambda x: "{}: {}".format(x[0], ', '.join(x[1] | map(str)))))
'Even: 0, 2, 4, 6, 8 / Odd: 1, 3, 5, 7, 9'
>>>

islice()

Just the itertools.islice function as a Pipe:

>>> from pipe import islice
>>> list((1, 2, 3, 4, 5, 6, 7, 8, 9) | islice(2, 8, 2))
[3, 5, 7]
>>>

izip()

Just the itertools.izip function as a Pipe:

>>> from pipe import izip
>>> list(range(0, 10) | izip(range(1, 11)))
[(0, 1), (1, 2), (2, 3), (3, 4), (4, 5), (5, 6), (6, 7), (7, 8), (8, 9), (9, 10)]
>>>

map(), select()

Apply a conversion expression given as parameter to each element of the given iterable

>>> list([1, 2, 3] | map(lambda x: x * x))
[1, 4, 9]

>>> list([1, 2, 3] | select(lambda x: x * x))
[1, 4, 9]
>>>

netcat

The netcat Pipe sends and receive bytes over TCP:

data = [
    b"HEAD / HTTP/1.0\r\n",
    b"Host: python.org\r\n",
    b"\r\n",
]
for packet in data | netcat("python.org", 80):
    print(packet.decode("UTF-8"))

Gives:

HTTP/1.1 301 Moved Permanently
Content-length: 0
Location: https://python.org/
Connection: close

permutations(r=None)

Returns all possible permutations:

>>> from pipe import permutations
>>> for item in 'ABC' | permutations(2):
...     print(item)
('A', 'B')
('A', 'C')
('B', 'A')
('B', 'C')
('C', 'A')
('C', 'B')
>>>

>>> for item in range(3) | permutations:
...     print(item)
(0, 1, 2)
(0, 2, 1)
(1, 0, 2)
(1, 2, 0)
(2, 0, 1)
(2, 1, 0)
>>>

reverse

Like Python's built-in reversed function.

>>> from pipe import reverse
>>> list([1, 2, 3] | reverse)
[3, 2, 1]
>>>

select(fct)

Alias for map(fct), see map(fct).

skip()

Skips the given quantity of elements from the given iterable, then yields

>>> from pipe import skip
>>> list((1, 2, 3, 4, 5) | skip(2))
[3, 4, 5]
>>>

skip_while(predicate)

Like itertools.dropwhile, skips elements of the given iterable while the predicate is true, then yields others:

>>> from pipe import skip_while
>>> list([1, 2, 3, 4] | skip_while(lambda x: x < 3))
[3, 4]
>>>

sort(key=None, reverse=False)

Like Python's built-in "sorted" primitive.

>>> from pipe import sort
>>> ''.join("python" | sort)
'hnopty'
>>> [5, -4, 3, -2, 1] | sort(key=abs)
[1, -2, 3, -4, 5]
>>>

t

Like Haskell's operator ":":

>>> from pipe import t
>>> for i in 0 | t(1) | t(2):
...     print(i)
0
1
2
>>>

tail(n)

Yields the given quantity of the last elements of the given iterable.

>>> from pipe import tail
>>> for i in (1, 2, 3, 4, 5) | tail(3):
...     print(i)
3
4
5
>>>

take(n)

Yields the given quantity of elements from the given iterable, like head in shell script.

>>> from pipe import take
>>> for i in count() | take(5):
...     print(i)
0
1
2
3
4
>>>

take_while(predicate)

Like itertools.takewhile, yields elements of the given iterable while the predicate is true:

>>> from pipe import take_while
>>> for i in count() | take_while(lambda x: x ** 2 < 100):
...     print(i)
0
1
2
3
4
5
6
7
8
9
>>>

tee

tee outputs to the standard output and yield unchanged items, useful for debugging a pipe stage by stage:

>>> from pipe import tee
>>> sum(["1", "2", "3", "4", "5"] | tee | map(int) | tee)
'1'
1
'2'
2
'3'
3
'4'
4
'5'
5
15
>>>

The 15 at the end is the sum returning.

transpose()

Transposes the rows and columns of a matrix.

>>> from pipe import transpose
>>> [[1, 2, 3], [4, 5, 6], [7, 8, 9]] | transpose
[(1, 4, 7), (2, 5, 8), (3, 6, 9)]
>>>

traverse

Recursively unfold iterables:

>>> list([[1, 2], [[[3], [[4]]], [5]]] | traverse)
[1, 2, 3, 4, 5]
>>> squares = (i * i for i in range(3))
>>> list([[0, 1, 2], squares] | traverse)
[0, 1, 2, 0, 1, 4]
>>>

uniq(key=None)

Like dedup() but only deduplicate consecutive values, using the given key function if provided (or else the identity).

>>> from pipe import uniq
>>> list([1, 1, 2, 2, 3, 3, 1, 2, 3] | uniq)
[1, 2, 3, 1, 2, 3]
>>> list([1, -1, 1, 2, -2, 2, 3, 3, 1, 2, 3] | uniq(key=abs))
[1, 2, 3, 1, 2, 3]
>>>

where(predicate), filter(predicate)

Only yields the matching items of the given iterable:

>>> list([1, 2, 3] | where(lambda x: x % 2 == 0))
[2]
>>>

Don't forget they can be aliased:

>>> positive = where(lambda x: x > 0)
>>> negative = where(lambda x: x < 0)
>>> sum([-10, -5, 0, 5, 10] | positive)
15
>>> sum([-10, -5, 0, 5, 10] | negative)
-15
>>>

Constructing your own

You can construct your pipes using the Pipe class like:

from pipe import Pipe
square = Pipe(lambda iterable: (x ** 2 for x in iterable))
map = Pipe(lambda iterable, fct: builtins.map(fct, iterable)
>>>

As you can see it's often very short to write, and with a bit of luck the function you're wrapping already takes an iterable as the first argument, making the wrapping straight forward:

>>> from collections import deque
>>> from pipe import Pipe
>>> end = Pipe(deque)
>>>

and that's it itrable | end(3) is deque(iterable, 3):

>>> list(range(100) | end(3))
[97, 98, 99]
>>>

In case it gets more complicated one can use Pipe as a decorator to a function taking an iterable as the first argument, and any other optional arguments after:

>>> from statistics import mean

>>> @Pipe
... def running_average(iterable, width):
...     items = deque(maxlen=width)
...     for item in iterable:
...         items.append(item)
...         yield mean(items)

>>> list(range(20) | running_average(width=2))
[0, 0.5, 1.5, 2.5, 3.5, 4.5, 5.5, 6.5, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5, 14.5, 15.5, 16.5, 17.5, 18.5]
>>> list(range(20) | running_average(width=10))
[0, 0.5, 1, 1.5, 2, 2.5, 3, 3.5, 4, 4.5, 5.5, 6.5, 7.5, 8.5, 9.5, 10.5, 11.5, 12.5, 13.5, 14.5]
>>>

Organizing pipes more effectively using classes

The @Pipe decorator isn't just for functions-it also works with classes. You can use it with instance methods, class methods, and static methods to better structure your code while keeping it pipeable.

1. Using an Instance Method