pipe

Pipe enables a shell-like infix syntax in Python using the | operator. This allows you to chain operations on iterables in a readable, linear fashion.

Each pipe is lazily evaluated, can be aliased (pre-configured with arguments), and can be partially initialized. This makes it ideal for building complex data processing pipelines.

Example of aliasing a pipe:

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

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

Pipe is built on generators, making it naturally lazy. Defining a pipe chain does not trigger any computation or data generation; values are only processed when a consumer (like list(), sum(), or a for loop) pulls them through the chain.

This is particularly useful when working with infinite generators like itertools.count(). The pipeline will only execute as many steps as necessary to satisfy the final consumer.

Example of lazy execution:

from itertools import count

# This chain defines the logic but executes nothing
result = (count()
          | tee
          | select(lambda x: x ** 2)
          | take(10)
          | where(lambda x: x % 2))

# Computation only starts here when we pull the first value
print(list(result | take(1)))

To install the pipe library, use pip via the command line. Use the specific command for your operating system:

Linux/macOS:

python3 -m pip install pipe

Windows:

py -3 -m pip install pipe

# Linux/macOS
python3 -m pip install pipe

# Windows
py -3 -m pip install pipe

In Pipe 2.0, all 'closing pipes' (pipes that return non-iterables, such as add or as_list) have been removed to ensure all pipes return iterables and maintain consistency.

If you are upgrading from 1.x and encounter exceptions, use one of these three strategies:

1. Replace closing pipes with standard Python functions: Instead of ... | as_list, use list(...). Instead of ... | add, use sum(...).
2. Reimplement necessary closing pipes: Since @Pipe makes it easy to create custom logic, you can implement the specific closing pipes you need in a few lines of code.
3. Pin the version: If you cannot refactor immediately, install the older version: pip install pipe<2.

Tip: Use Python's Development Mode to catch deprecation warnings before they become breaking changes.

The tee pipe is useful for inspecting the state of a pipeline at any given stage. It prints the current items to standard output and then yields them unchanged to the next stage in the pipe.

from pipe import tee, map

# This will print '1', '2', '3', '4', '5' to stdout during execution
sum(["1", "2", "3", "4", "5"] | tee | map(int) | tee)

You can define your own infix operations by decorating a generator function with @Pipe. A decorated function must accept an iterable as its first argument and yield values. This allows you to use your custom function within a pipe chain using the | operator.

For multi-argument pipes, you can partially initialize them (currying) to create specialized pipes. This works with both positional and keyword arguments.

Example of a custom pipe with partial initialization:

@Pipe
def addmul(iterable, to_add, to_mul):
    """Computes (x + to_add) * to_mul to every items of the input."""
    for i in iterable:
        yield (i + to_add) * to_mul

# Partially initialize with to_add=0
add_zero = addmul(0)
# Use the specialized pipe
list(range(10) | add_zero(10))

You can create your own pipes by instantiating the Pipe class. The class takes a function that accepts an iterable as its first argument.

Basic Pipe

from pipe import Pipe
square = Pipe(lambda iterable: (x ** 2 for x in iterable))

Wrapping existing functions

If a function already takes an iterable as its first argument, wrapping it is straightforward:

from collections import deque
from pipe import Pipe
end = Pipe(deque)
# Usage: list(range(10) | end(3))

One-off pipes with arguments

You can specify positional and named arguments directly when creating a Pipe instance:

from itertools import combinations
list(range(5) | Pipe(combinations, 2))