Your service writes a 20 GB log file, and you want to count the lines that
contain ERROR. f.readlines() builds a list of every line, so the whole file
must fit in memory first. Pass the lines through a generator instead:
from collections.abc import Iterable, Iterator
def error_lines(lines: Iterable[str]) -> Iterator[str]:
for line in lines:
if "ERROR" in line:
yield line
log = ["INFO start\n", "ERROR disk full\n", "INFO retry\n", "ERROR disk full\n"]
print(sum(1 for _ in error_lines(log))) # 2
The list log stands in for the file. A for loop over an open file f also
reads one line at a time. So counting with error_lines(f) keeps about one
line in memory, however big the file is. sum(1 for _ in ...) counts the lines without
building a list.
You wrote a generator like error_lines in
Generators and One-Shot Iterators,
and this track builds on it.
A list holds every value, a generator holds one
This script sums 100,000 squares from a list, then from a generator expression, and measures the peak memory of each:
import tracemalloc
from collections.abc import Callable
def eager_total() -> int:
values = [n * n for n in range(100_000)]
return sum(values)
def lazy_total() -> int:
return sum(n * n for n in range(100_000))
def measure(work: Callable[[], int]) -> tuple[int, int]:
tracemalloc.start()
try:
result = work()
_, peak = tracemalloc.get_traced_memory()
return result, peak
finally:
tracemalloc.stop()
eager_result, eager_peak = measure(eager_total)
lazy_result, lazy_peak = measure(lazy_total)
assert eager_result == lazy_result
print({"eager_peak_bytes": eager_peak, "lazy_peak_bytes": lazy_peak})
tracemalloc tracks the memory Python allocates for objects, and records the
highest total, the peak. On Python 3.12 the list version peaks at about 4 MB,
because it stores all 100,000 numbers before sum starts. The generator version peaks at under 1 KB. It makes one
square, hands it to sum, and drops it before it makes the next. Make the
range ten times bigger, and the list's peak grows to about 40 MB while the
generator's stays the same.
Choose when an argument check runs
Where you put an argument check decides when it runs. Compare these two:
from collections.abc import Iterable, Iterator
from itertools import batched
def pages_checked_on_next(lines: Iterable[str], size: int) -> Iterator[list[str]]:
if size <= 0:
raise ValueError(f"size must be positive, got {size}")
for batch in batched(lines, size):
yield list(batch)
def pages_checked_on_call(lines: Iterable[str], size: int) -> Iterator[list[str]]:
if size <= 0:
raise ValueError(f"size must be positive, got {size}")
return (list(batch) for batch in batched(lines, size))
pages = pages_checked_on_next(["a"], 0) # no error yet
try:
next(pages)
except ValueError as exc:
print("at the first next():", exc)
try:
pages_checked_on_call(["a"], 0)
except ValueError as exc:
print("at the call:", exc)
pages_checked_on_next has yield in its body, so it is a generator
function. Its check runs with the rest of the body, at the first next().
pages_checked_on_call has no yield. It is a plain function that checks the
size and then returns a generator, so its check runs during the call. You
wrote this second style in bulk_pages.
In a library, check at the call. The traceback then points at the line that
passed the bad size, not at a loop somewhere else.
The batching exercise later in this track
asks for the check inside the body on purpose. Its driver shows that nothing
in the body runs before the first next(), and that includes the check.
Calling list() on a generator stores everything again
list(error_lines(f)) puts every error line in memory, and the saving is
gone. Loop over the generator instead, or pass it to a function that takes
one value at a time, such as sum, max or any.
A list is the right choice when the caller needs len(), indexing or more
than one pass, and the data fits in memory.
A generator reads variables when it runs
A generator looks up its variables when its body runs, not when you create it:
threshold = 10
big = (n for n in [5, 15, 25] if n > threshold)
threshold = 20
print(list(big)) # [25], not [15, 25]
The test n > threshold runs during list(big), after threshold became 20.
To use the value from when you create the generator, pass it as an argument
to a generator function. The function gets its arguments at the call, even though
its body runs later.
Close the file when you stop early
A generator can pause inside a with block. Here read_rows pauses at each
yield with the file still open:
from collections.abc import Generator
from contextlib import closing
def read_rows(path: str) -> Generator[str, None, None]:
with open(path, encoding="utf-8") as handle:
yield from handle
def first_data_row(path: str) -> str | None:
with closing(read_rows(path)) as rows:
for row in rows:
if not row.startswith("#"):
return row.rstrip("\n")
return None
yield from handle hands out the file's lines one by one. When
first_data_row returns early, read_rows is still paused, with the file
open. Without closing, the file stays open until Python frees the generator,
and that depends on what else refers to it. closing() calls the generator's
close() method when the with closing(...) block ends. That finishes the
generator at the paused yield, so the generator's own with block closes the
file.
Or let the caller open the file and pass it in, as error_lines does above.
Then the caller's with block closes it.
Check your understanding
You call pages_checked_on_next(lines, 0) and never loop over the result. Does anything raise?
Show answer
Hide answer
No. The check is in the generator body, and the body runs only at the first
next(). With no loop, nothing in the body runs, so the bad size goes
unnoticed. For an error at the call, write the function like
pages_checked_on_call.
Going further
read_rows returns Generator[str, None, None], not Iterator[str], because
closing() needs a close() method and only the Generator type promises
one. The article on iterator types
explains its three type arguments.