lazy pipeline object

stream.py

from typing import Iterable
import itertools


class Stream:
    def __init__(self, stream: Iterable):
        self.stream = stream

    def __iter__(self):
        yield from self.stream

    def add_a(self)->"Stream":
        def inner():
            for x in self.stream:
                print(" adding a")
                yield x+"a"
        return Stream(inner())

    def add_b(self)->"Stream":
        def inner():
            for x in self.stream:
                print(" adding b")
                yield x+"b"
        return Stream(inner())

    def duplicate(self)->"Stream":
        def inner():
            print(" first")
            second = []
            for x in self.stream:
                yield x
                second.append(x)
            print(" second")
            for x in second:
                yield x
        return Stream(inner())

    def elem_type(self):
        try:
            tmp = next(self.stream)
        except TypeError:
            tmp = next(iter(self.stream))
        else:
            self.stream = itertools.chain([tmp], self.stream)
        return type(tmp)

usage.py

# lazily exhausts the complete stream without aggregating between calls -- pipeline bugs are discovered quickly
>>> for element in Stream("12345").add_a().add_b().duplicate():
...     print(element)
 first
 adding a
 adding b
1ab
 adding a
 adding b
2ab
 adding a
 adding b
3ab
 adding a
 adding b
4ab
 adding a
 adding b
5ab
 second
1ab
2ab
3ab
4ab
5ab

# operation that requires only one fully processed element stops without processing all data
>>> Stream("12345").add_a().add_b().duplicate().elem_type()
 first
 adding a
 adding b
<class 'str'>