Python Generators

What Is a Generator?

A generator function uses yield to produce values one at a time. Each call to next() resumes execution until the next yield.

def countdown(n):
    print("Starting countdown")
    while n > 0:
        yield n
        n -= 1
    print("Done")

gen = countdown(3)
next(gen)   # prints "Starting countdown", returns 3
next(gen)   # returns 2
next(gen)   # returns 1
next(gen)   # prints "Done", raises StopIteration

# Using in a for loop (handles StopIteration automatically)
for n in countdown(5):
    print(n)   # 5, 4, 3, 2, 1

Generator Expressions

Like list comprehensions but lazy — no parentheses around the iterable:

# List comprehension — builds entire list
squares_list = [x**2 for x in range(1_000_000)]   # ~8 MB

# Generator expression — computes on demand
squares_gen = (x**2 for x in range(1_000_000))    # ~120 bytes

# Use directly in function calls (one pair of parens needed)
total = sum(x**2 for x in range(1_000_000))
maximum = max(len(line) for line in open("large_file.txt"))

Infinite Sequences

Generators can produce infinite sequences — safe because values are computed on demand.

def integers(start=0):
    n = start
    while True:
        yield n
        n += 1

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# Take first N values with itertools.islice
from itertools import islice

first_10_fibs = list(islice(fibonacci(), 10))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

Pipelines with Generators

Chain generators together for memory-efficient data processing:

import csv
from pathlib import Path

def read_lines(path):
    """Yield lines from a large file without loading it all."""
    with open(path, encoding="utf-8") as f:
        yield from f

def parse_csv_rows(lines):
    """Parse CSV rows from a line stream."""
    reader = csv.DictReader(lines)
    yield from reader

def filter_active(rows):
    """Keep only active users."""
    for row in rows:
        if row["status"] == "active":
            yield row

def extract_emails(rows):
    """Extract the email field."""
    for row in rows:
        yield row["email"].strip().lower()

# Build the pipeline — nothing executes until iteration
pipeline = extract_emails(
    filter_active(
        parse_csv_rows(
            read_lines("users.csv")
        )
    )
)

# Process the file line by line — constant memory regardless of file size
for email in pipeline:
    send_newsletter(email)

yield from

yield from delegates to a sub-generator:

def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)   # recurse
        else:
            yield item

list(flatten([1, [2, [3, 4]], [5, 6]]))
# [1, 2, 3, 4, 5, 6]

# Also works for delegating to any iterable
def chain_iterables(*iterables):
    for it in iterables:
        yield from it

list(chain_iterables([1, 2], [3, 4], [5]))   # [1, 2, 3, 4, 5]

send() and Two-Way Communication

def accumulator():
    total = 0
    while True:
        value = yield total   # yield sends total out, receives value in
        if value is None:
            break
        total += value

gen = accumulator()
next(gen)          # prime the generator (advance to first yield), returns 0
gen.send(10)       # returns 10
gen.send(20)       # returns 30
gen.send(5)        # returns 35
gen.close()        # raises GeneratorExit inside the generator

itertools — The Generator Toolkit

import itertools

# Infinite iterators
itertools.count(10, 2)          # 10, 12, 14, 16, ...
itertools.cycle([1, 2, 3])      # 1, 2, 3, 1, 2, 3, ...
itertools.repeat("x", 3)        # "x", "x", "x"

# Combinatorics
list(itertools.permutations("ABC", 2))
# [('A','B'), ('A','C'), ('B','A'), ('B','C'), ('C','A'), ('C','B')]

list(itertools.combinations("ABC", 2))
# [('A','B'), ('A','C'), ('B','C')]

list(itertools.combinations_with_replacement("AB", 2))
# [('A','A'), ('A','B'), ('B','B')]

# Chaining
list(itertools.chain([1, 2], [3, 4], [5]))     # [1, 2, 3, 4, 5]
list(itertools.chain.from_iterable([[1,2],[3,4]]))  # [1, 2, 3, 4]

# Slicing infinite iterators
list(itertools.islice(itertools.count(), 5))   # [0, 1, 2, 3, 4]

# Grouping consecutive elements
data = [("A", 1), ("A", 2), ("B", 3), ("B", 4), ("A", 5)]
for key, group in itertools.groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# A [('A',1), ('A',2)]
# B [('B',3), ('B',4)]
# A [('A',5)]

# Zip with fill
list(itertools.zip_longest([1,2,3], ["a","b"], fillvalue=None))
# [(1,'a'), (2,'b'), (3,None)]

# Accumulate
list(itertools.accumulate([1, 2, 3, 4, 5]))            # [1, 3, 6, 10, 15]
list(itertools.accumulate([1, 2, 3, 4], func=max))     # [1, 2, 3, 4]

# Batched (Python 3.12+)
list(itertools.batched([1,2,3,4,5,6,7], 3))
# [(1,2,3), (4,5,6), (7,)]

Generator-Based Context Manager

from contextlib import contextmanager
import time

@contextmanager
def timer(label=""):
    start = time.perf_counter()
    try:
        yield   # control passes to the with block
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.4f}s")

with timer("list comprehension"):
    result = [x**2 for x in range(1_000_000)]

with timer("generator sum"):
    result = sum(x**2 for x in range(1_000_000))

Practical Example: Streaming Data Processing

import json
from typing import Generator

def stream_json_lines(path: str) -> Generator[dict, None, None]:
    """Parse a large JSON Lines file without loading it all."""
    with open(path) as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)

def filter_errors(events):
    return (e for e in events if e.get("level") == "error")

def extract_message(events):
    return (e["message"] for e in events)

def deduplicate(items):
    seen = set()
    for item in items:
        if item not in seen:
            seen.add(item)
            yield item

# Process a 10GB log file with constant memory
errors = deduplicate(
    extract_message(
        filter_errors(
            stream_json_lines("app.log.jsonl")
        )
    )
)

for message in errors:
    print(message)