Pipeline Demo

pipeline.py

class Job():
    def __init__(self, *args, **kwargs):
        pass

    def __call__(self, data):
        raise NotImplementedError()

class Square(Job):
    def __call__(self, data):
        return [d * d for d in data]

class Add(Job):
    def __init__(self, addend, *args, **kwargs):
        self.addend = addend

    def __call__(self, data):
        return [d + self.addend for d in data]


def job_lambda(job):
    def f(*args, **kwargs):
        return job(*args, **kwargs)

    return f


class Pipeline():
    def __init__(self):
        self.jobs = []

    def __getattr__(self, name):
        def middleman(job):
            def f(*args, **kwargs):
                self.jobs.append(job(*args, **kwargs))
                return self
            return f
        available_transforms = {
            'Square': Square,
            'Add': Add,
        }
        return middleman(available_transforms[name])

    def __call__(self, data):
        data_copy = list(data)
        for job in self.jobs:
            data_copy = job(data_copy)
        return data_copy

if __name__ == '__main__':
    p = Pipeline().Square().Add(5)
    p.Add(2)

    sequence1 = list(range(5))
    sequence2 = list(range(0, 10, 2))

    print(f'{sequence1} -> {p(sequence1})
    print(f'{sequence2} -> {p(sequence2})

Example usage:

$ python pipeline.py
[0, 1, 2, 3, 4] -> [7, 8, 11, 16, 23]
[0, 2, 4, 6, 8] -> [7, 11, 23, 43, 71]