Streaming pipelines end to end

Individually, lazy stages are easy. Composed, they expose the real skill: knowing which stage breaks the streaming property.

sorted() is the canonical offender. itertools.groupby only groups adjacent equal keys, so the obvious way to make it work is groupby(sorted(rows, key=...)) — and that single call materialises the entire stream. A pipeline with a sorted() in the middle is not a streaming pipeline, and a learner who writes one and then claims “fully streaming” has not understood the contract. (The honest fixes are: require the input pre-sorted, or accept that you have a batch job.)

This pipeline avoids the trap by grouping within a batch — every stage is lazy except the aggregate, which is allowed to materialise because it only ever sees batch items.

What to write

Four stages plus the driver. The signatures are the exercise as much as the bodies:

def flatten(sources: Iterable[Iterable[str]]) -> Iterator[str]
def parse(lines: Iterable[str]) -> Iterator[tuple[str, int]]
def aggregate(pairs: Iterable[tuple[str, int]]) -> dict[str, int]
def process(sources: Iterable[Iterable[str]], *, batch: int, limit: int) -> Iterator[dict[str, int]]

Note the asymmetry, and copy it: every stage accepts Iterable (the broadest thing it can honestly work with) and returns Iterator (the narrowest honest description of what it hands back). Accept broad, return narrow. Annotating a stage’s return as Iterable throws away the information that the result is single-shot; annotating a parameter as Iterator refuses lists for no reason.

• flatten concatenates the inner iterables lazily. itertools.chain.from_iterable is exactly this.
• parse turns "key=value" into (key, int(value)).
• aggregate sums values per key and returns a plain dict. This is the only stage permitted to materialise.
• process composes them: flatten, cap at limit lines, group into batches of batch pairs, aggregate each batch.

itertools.batched (new in 3.12) yields tuples lazily and is the right tool for the batching stage.

The placement question

islice(..., limit) must sit after flattening and before batching. Put it before the flatten and it caps the number of sources; put it after the batching and it caps the number of batches; leave it out and the infinite test never terminates. There is exactly one correct position and the tests find it.

What the report proves

• batches — the per-batch aggregates, in order.
• pulled_before_first — must be 0, because process is a generator function and calling it runs none of its body.
• pulled — total lines read. With an infinite outer source and limit=1000 it must be exactly 1000: not one line more, and the run must finish.

  Loading visualization…