Skip to content

← Laziness, Iteration and Pipelines step 13 of 14

Medium Primitives

The single-consumption contract

Pass the same generator to two functions and the second one gets nothing. No exception. No warning. Just an empty result that looks like a legitimate empty result. It is the quietest bug in the iteration protocol, and it exists because an iterator is its own __iter__ — so it satisfies Iterable[T] perfectly while being single-shot.

itertools.tee is the tempting answer and usually the wrong one. tee buffers everything a lagging iterator has not yet consumed, so if one copy is drained before the other starts, tee‘s internal buffer holds the whole stream anyway — with more indirection and more objects than a list. The stdlib documentation says so outright: if one iterator will use most of the data before another starts, it is faster to use list().

So build the honest version: materialise once, bounded, and hand back independent copies.

What to write

def tee_safely[T](source: Iterable[T], n: int, *, cap: int) -> tuple[list[T], ...]
  • If n < 0, raise ValueError(f"n must be non-negative, got {n}") before touching the source. The report’s iter_calls is 0 in that case, which is how the test knows you validated first.
  • Otherwise pull at most cap + 1 items. If more than cap arrive, raise ValueError(f"source exceeds the {cap}-item cap"). Pulling exactly one item past the cap is the whole trick: one extra item is precisely enough evidence that the limit was exceeded, and not one item more. A solution that materialises everything and then checks len has already lost.
  • Return a tuple of n independent lists. Mutating one must not affect another, and n == 0 returns the empty tuple.
  • Iterate source exactly once, whatever n is.
def solve(*, items, n, cap, lazy=False) -> Report

Build the source (lazy=True wraps items in a generator so it is genuinely single-shot), wrap it in CountingSource, call tee_safely, and report copies, independent, iter_calls, pulled and error. _independent is provided.

The container type is part of the answer

The harness distinguishes a tuple from a list. return [list(buf) for _ in range(n)] fails even with byte-identical contents. That strictness is deliberate: tuple[list[T], ...] says “a fixed bundle of results” and list[list[T]] says “a sequence you may append to”, and the difference shows up the first time a caller does left, right = tee_safely(src, 2, cap=1000).

Loading visualization…