Provides a simple API for running an object through a sequence of computations ("pipes"). Think of Unix pipelines. Works with async pipes!

Inspired by the pipeline implementation of the Laravel framework.

Copy pipeline.py to your project.

Here is a basic example.

pipe('desserts').through([
    Reverse,
    lambda s, next: next(s.upper()),
    Wrap('__')
]).run()  # '__STRESSED__'

Pipes are provided as a list and can be function or class based. Class based pipes have to have a handle method. Both function based pipes and the handle method on class based pipes receive two parameters: the data from the previous pipe and a next-function, that has to be called to hand over to the next pipe. In the example above Reverse and Wrap are examples of class based pipes, while the middle pipe is provided as a lambda.

class Reverse(Pipe):
    def handle(self, content, next):
        return next(content[::-1])


class Wrap(Pipe):
    def __init__(self, wrapper_string: str):
        self.wrapper_string = wrapper_string

    def handle(self, content, next):
        return next(self.wrapper_string + content + self.wrapper_string)

More examples (including how to use with async pipes) at the bottom of pipeline.py.

See docstring on the Pipeline class in pipeline.py.

Run

python pipeline.py

from the command line. If you get some text output and no errors - it works.