Chaining Generators into a Pipeline
In short: Quarry never does just one thing to a log line — it reads, parses, filters, and stops once it has enough. The eager way runs each step over the whole input and hands the next step a list. The lazy way makes each step a generator that takes an iterable and yields, so the steps snap together like pipe sections: one record crosses every stage before the next is even read. Requests travel backward, from consumer to source; values travel forward, one at a time. That's what lets a lazy pipeline stop early and read only as much of a 200 MB log as the answer actually needs.
Chain four stages, and nothing runs yet
Every stage has the same shape: take an iterable, loop over it, yield. Because the input can be any iterable, it can be another stage:
Building the chain runs no code in any stage — that's why printing
errors shows only a generator object. Work starts when the for loop at
the bottom asks for the first record. The mistake people make here is
picturing the eager version, where each stage finishes before the next
begins:
def read(lines):
for line in lines:
print("r" + line, end=" ")
yield line
def shout(lines):
for line in lines:
print("s" + line, end=" ")
yield line.upper()
for x in shout(read(["a", "b"])):
print(x, end=" ")If you guessed ra rb sa sb A B, that's how a chain of lists behaves, with
a full list built between every step. A lazy pipeline never builds that
list — each stage holds one item at most, which is why its memory cost
doesn't depend on the size of the input. The two shapes side by side:
Stop a pipeline before it reads the whole log
The biggest win isn't memory — it's that a lazy pipeline can stop. Quarry's preview wants the first two errors from a file that might have ten million lines. Count how many lines actually leave the source:
pulled = 0
def source(n):
global pulled
for i in range(n):
pulled += 1
yield i
evens = (x for x in source(10**9) if x % 2 == 0)
first_two = next(evens), next(evens)
print(first_two, pulled)A take(items, n) stage is how a pipeline says "enough" — it must stop the
moment it has yielded its nth item, never pulling one more than needed:
Four lines pulled out of six. Put if count == n at the top of the loop
instead of after the yield, and take would pull one extra record before
deciding it has enough — with a slow remote source, that's one wasted fetch
every time. The standard library already has this stage: itertools.islice
does exactly what take does, with the same discipline about not
over-pulling, and itertools.chain joins several sources into one stream:
How far a lazy pipeline has to read depends on how common the thing you want is. At 10% errors, finding 20 of them reads about 200 lines; at 1%, about 2,000 — rarer matches mean reading further, but it still stops the moment it has enough. An eager pipeline reads every line regardless, so the same preview on a 10-million-line file costs 10 million reads either way.
Reuse a parsed stream for two different answers
Some operations — sorted(), counting how often each level appears — need
the whole stream before they can produce anything. Put those last,
after the filters have made the stream as small as possible; sorting 200
errors is cheap, sorting ten million lines first isn't.
Every stage is a generator, so every stage is single-use. That catches people the first time they try to get two different answers from one stream:
Run it and look at infos. Then change the code so both lists come out right, without reading lines into anything but the two answers.
There's no error, just a silently empty list — the dangerous kind of bug. If two consumers need the same stream, either give each one its own fresh pipeline built from a function, or split the records in a single pass.
Quick practice
nums = (n for n in range(5))
doubled = (n * 2 for n in nums)
print(next(doubled), next(doubled), list(nums))from itertools import islice
def forever():
n = 0
while True:
yield n
n += 1
print(list(islice(forever(), 3, 6)))🎯 Your turn
Build Quarry's log pipeline from four stages, and prove it's lazy.
counted(lines, tally): yields every line unchanged, adding 1 totally["pulled"]for each one it pulls (provided).parse(lines): skips blank lines and yields{"level": ..., "msg": ...}, splitting each line at its first space.at_level(records, level): yields only records at that level.take(items, n): yields at mostnitems and stops as soon as it has yielded thenth, without pulling another. Whennis 0 it yields nothing and pulls nothing.
Then pipe(lines, level, limit) chains them and returns
{"records": [msg, ...], "pulled": n}:
pipe(["INFO boot", "ERROR disk full", "INFO tick", "ERROR fan stuck", "ERROR late", "INFO bye"], "ERROR", 2)→{"records": ["disk full", "fan stuck"], "pulled": 4}
Hint: take checks n <= 0 before its loop and count == n after
each yield. Build the chain from the inside out: take(at_level(parse(counted(lines, tally)), level), limit).
Press ▶ Run, then ✓ Check. ✅