Python Generators: Build Streaming Pipelines Without Losing the Audit

Reviewed October 6, 2026. Examples use Python 3.12.14 and original synthetic records. This is a semantics tutorial, not a memory or speed benchmark.

A CSV analysis works on a hundred rows, so you replace the list comprehension with a generator expression before processing a much larger file. The program now appears more efficient. Its second report, however, contains zero rows. Elsewhere, a preview consumes the first records before the actual calculation begins.

The missing ingredient is an ownership model for iteration. A generator is an iterator that advances through a computation; it is not a collection that automatically remembers every value. Laziness can reduce retained intermediate results, but it also changes when errors happen, which code owns the source, and whether a second pass is possible.

Practical rule: design the whole pipeline around a clearly stated number of passes. Compute related aggregates together, preserve an exclusion audit, and identify every operation that materializes or buffers data. Use a fresh source when a genuinely independent second pass is required.

Iterable, iterator, and generator are different roles

A list is iterable: calling iter on it creates an iterator. An iterator supplies values through next and eventually signals StopIteration. A generator function uses yield to produce values while suspending its own execution state. Calling that function creates a generator object; the function body starts advancing when the consumer requests values.

An ordinary list can produce a new iterator for another pass. Calling iter on an existing iterator normally returns that same iterator, at its current position. Wrapping an exhausted generator in a for loop therefore does not restart its source or rebuild its values.

The Python expression reference also makes an important distinction for generator expressions: the leftmost iterable expression is evaluated immediately, while later work is deferred. Do not describe every part of a generator expression as delayed without qualification.

values = [10, 20, 30]
stream = (value * 2 for value in values)
assert next(stream) == 20
assert list(stream) == [40, 60]
assert list(stream) == []
assert list(value * 2 for value in values) == [20, 40, 60]
print("A new generator gives a new pass; the old one stays exhausted")

The final expression creates a new object using a repeatable list source. That would not necessarily replay a network response, database cursor, or another exhausted iterator. Reusability is a property of the source and construction, not the parentheses alone.

A single pass can calculate more than one statistic

Suppose a field contains whole minutes. An empty field means absent, zero is a valid measurement, and other text may be malformed. The report needs accepted count, rejected count, total minutes, and mean minutes. Running four independent consumers against one generator is the wrong design.

Instead, classify each row once and update related accumulators together. This keeps the denominator aligned with the accepted values. It also lets you reconcile the source count with accepted plus rejected before presenting any result.

from io import StringIO
import csv

text = "topic,minutes\nfunctions,30\nreview,0\nfiles,\nloops,oops\n"

def classified_rows(file):
    for number, row in enumerate(csv.DictReader(file), start=2):
        raw = row["minutes"].strip()
        if not raw:
            yield number, None, "missing minutes"
            continue
        try:
            minutes = int(raw)
        except ValueError:
            yield number, None, "invalid integer"
            continue
        if not 0 <= minutes <= 240:
            yield number, None, "minutes outside range"
            continue
        yield number, minutes, None

def summarize(classified):
    seen = accepted = total = 0
    rejected = []
    for number, minutes, reason in classified:
        seen += 1
        if reason is not None:
            rejected.append({"line": number, "reason": reason})
        else:
            accepted += 1
            total += minutes
    assert seen == accepted + len(rejected)
    return {"seen": seen, "accepted": accepted, "rejected": rejected,
            "total": total, "mean": total / accepted if accepted else None}

with StringIO(text) as file:
    result = summarize(classified_rows(file))
assert result["accepted"] == 2 and result["total"] == 30
assert result["mean"] == 15
assert [item["line"] for item in result["rejected"]] == [4, 5]
print(result)

The reported 15 minutes is 30 divided by two accepted records. It is not the mean over every source row and not the mean over only positive values. The generator changes delivery; it does not choose the statistical population for you.

Streaming does not automatically mean constant memory

The example retains every rejection in a list. If almost every input fails, that audit can grow with the source. The input text is also already fully present in memory. This fixture demonstrates control flow and semantics, not the memory behavior of a real file reader.

For a large production source, you might write rejection records incrementally to an audit file while retaining only counts and totals. You would then need to manage both resources, handle failed writes, and ensure the report identifies the audit artifact. A bounded recent-error sample is another policy, but it cannot replace a complete audit if the task requires one.

The itertools documentation specifically explains tee's auxiliary storage tradeoff. Two apparently independent iterators are not a free way to keep one consumer arbitrarily far behind another. A single summary pass is often clearer when the calculations share the same population.

Keep the source open until consumption finishes

A generator can outlive the file it is supposed to read. Returning a reader from inside a with block and consuming it afterward closes the file too early. The source boundary must own the resource for the entire period in which the consumer advances it.

One pattern is to consume inside the caller's with block, as the example does. Another is a generator function whose own with block spans the yields. In the latter case, early termination needs a deliberate cleanup policy rather than relying on an unspecified future garbage collection event. Explicitly closing a generator is useful when it owns resources and the consumer stops early.

Errors can also occur during consumption rather than object construction. Put conversion handling around the actual operation, and do not assume that successfully creating a generator proves the stream can be fully processed. A useful test requests the first value, a middle value, and exhaustion under inputs that fail at different positions.

A verification plan for the pipeline

Before changing the implementation, record expected behavior for an empty file, all-invalid rows, a valid zero, a negative value, a value above the range, and a malformed field. Verify that each source record receives exactly one classification. Assert the final denominator and preserve source line numbers in the audit.

Then test consumer behavior: a preview consumes values; a second pass needs a newly constructed source; materialization returns only values remaining at that point. If you use a live source, document whether reopening it produces the same snapshot. A changing file or API can make a second pass a different dataset even when the code is identical.

The point of generators is controlled delivery with an explicit state model. The reliable pipeline makes its consumption, retained state, resource lifetime, and exclusions visible. Measure memory and elapsed time separately on representative workloads before claiming a performance improvement.

Sources and reproducibility

Primary references: Python generator expressions and yield, Python iterator terminology, and itertools, including tee. Both examples are self-contained and use no network services. All values are synthetic; the assertions establish the reported results.