Skip to main content
Advertisement

Generator Functions and Expressions — yield, send(), throw()

A generator is a special function that uses the yield keyword to produce values one at a time. Generators simplify iterator creation, save memory, and form the foundation of coroutines.

The yield Keyword: Generator Functions​

A function that contains yield in its body is a generator function. Calling it returns a generator object.

def count_up(start: int, stop: int):
    """Looks like a regular function, but yield makes it a generator function"""
    current = start
    while current <= stop:
        yield current   # Return value + pause
        current += 1    # Resumes here when next() is called

# Calling it returns a generator object (code does NOT execute yet)
gen = count_up(1, 5)
print(type(gen))   # <class 'generator'>
print(gen)         # <generator object count_up at 0x...>

# Pull values with next()
print(next(gen))   # 1
print(next(gen))   # 2

# Consume with a for loop
for n in count_up(1, 5):
    print(n)   # 1, 2, 3, 4, 5

Regular Function vs Generator Function​

# Regular function: loads the entire list into memory
def squares_list(n: int) -> list[int]:
    return [x ** 2 for x in range(n)]

# Generator: produces one at a time — memory O(1)
def squares_gen(n: int):
    for x in range(n):
        yield x ** 2

import sys
print(sys.getsizeof(squares_list(1_000_000)))  # ~8MB
gen = squares_gen(1_000_000)
print(sys.getsizeof(gen))                       # ~112 bytes

Preserving Function State with yield​

def stateful_counter():
    """yield preserves the function's state (local variables, execution position)"""
    count = 0
    while True:
        received = yield count   # Return value and pause
        if received == "reset":
            count = 0
        else:
            count += 1

counter = stateful_counter()
print(next(counter))    # 0 (initialize)
print(next(counter))    # 1
print(next(counter))    # 2
counter.send("reset")   # Reset
print(next(counter))    # 1

Generator Expressions​

Same syntax as list comprehensions but uses () instead of [].

# List comprehension: eagerly evaluated, stored in memory
squares_list = [x**2 for x in range(10)]

# Generator expression: lazily evaluated, memory-efficient
squares_gen = (x**2 for x in range(10))

print(type(squares_list))  # <class 'list'>
print(type(squares_gen))   # <class 'generator'>

# Consuming
print(sum(x**2 for x in range(10)))         # 285 — parentheses can be omitted
print(list(x**2 for x in range(10)))        # [0, 1, 4, 9, ...]
print(max(len(word) for word in ["hi", "hello", "hey"]))  # 5
# Generator expression with condition
evens = (x for x in range(20) if x % 2 == 0)
print(list(evens))  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

# Nested generator
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flat = (elem for row in matrix for elem in row)
print(list(flat))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

send(): Sending Values to a Generator​

send(value) resumes the generator like next(), while also passing the value as the result of the yield expression.

def accumulator():
    """Receives values from the outside and accumulates them"""
    total = 0
    while True:
        value = yield total   # Return total, store received value in value
        if value is None:
            break
        total += value


gen = accumulator()
next(gen)             # Start generator (run to first yield), returns 0
print(gen.send(10))   # Send 10 → total=10 → return 10
print(gen.send(20))   # Send 20 → total=30 → return 30
print(gen.send(5))    # Send 5 → total=35 → return 35
gen.send(None)        # Termination signal

Note: Before the first send() call, you must start the generator with next() or send(None).

# Priming decorator for automatic startup
from functools import wraps
from typing import Callable, Any

def prime(gen_func: Callable) -> Callable:
    """Decorator that automatically starts a generator function"""
    @wraps(gen_func)
    def wrapper(*args, **kwargs):
        gen = gen_func(*args, **kwargs)
        next(gen)  # Auto-run to first yield
        return gen
    return wrapper


@prime
def running_average():
    """Real-time average calculator"""
    total, count = 0, 0
    while True:
        value = yield total / count if count > 0 else 0
        total += value
        count += 1


avg = running_average()   # Automatically primed
print(avg.send(10))  # 10.0
print(avg.send(20))  # 15.0
print(avg.send(30))  # 20.0

throw(): Injecting Exceptions​

Throws an exception into the generator from the outside.

def generator_with_error_handling():
    """Generator that handles exceptions injected from outside"""
    value = 0
    while True:
        try:
            value = yield value
        except ValueError as e:
            print(f"ValueError caught: {e}")
            value = 0  # Reset
        except GeneratorExit:
            print("Generator close requested")
            return


gen = generator_with_error_handling()
next(gen)                       # Start
print(gen.send(10))             # 10
print(gen.send(20))             # 20
gen.throw(ValueError, "bad value")  # Inject ValueError → reset to 0
print(next(gen))                # 1 (0+1)
gen.close()                     # GeneratorExit raised

close(): Terminating a Generator​

close() throws a GeneratorExit exception into the generator to terminate it gracefully.

def resource_holding_gen():
    """Generator that holds a resource"""
    print("Resource acquired")
    try:
        while True:
            yield "data"
    finally:
        print("Resource released")  # Always runs when close() is called


gen = resource_holding_gen()
print(next(gen))   # "data"
gen.close()        # GeneratorExit → finally runs → "Resource released"

Infinite Sequences and Lazy Evaluation​

def naturals(start: int = 1):
    """Infinite natural number sequence"""
    n = start
    while True:
        yield n
        n += 1

def primes():
    """Sieve of Eratosthenes — infinite prime sequence"""
    composites: set[int] = set()
    for n in naturals(2):
        if n not in composites:
            yield n
            composites.update(range(n * n, n * n + n * 100, n))

import itertools

# First 10 primes
first_10_primes = list(itertools.islice(primes(), 10))
print(first_10_primes)  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

# Primes under 100
small_primes = list(itertools.takewhile(lambda x: x < 100, primes()))
print(small_primes)

Practical Example: Data Pipeline​

from pathlib import Path
import csv
from typing import Iterator

def read_csv_rows(filepath: str | Path) -> Iterator[dict]:
    """Yield CSV file rows one at a time"""
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield dict(row)

def filter_active(rows: Iterator[dict]) -> Iterator[dict]:
    """Filter only active users"""
    for row in rows:
        if row.get("status") == "active":
            yield row

def enrich_data(rows: Iterator[dict]) -> Iterator[dict]:
    """Transform/enrich data"""
    for row in rows:
        row["full_name"] = f"{row['first_name']} {row['last_name']}"
        row["age"] = int(row.get("age", 0))
        yield row

def batch_rows(rows: Iterator[dict], size: int = 100) -> Iterator[list[dict]]:
    """Group into batches"""
    batch = []
    for row in rows:
        batch.append(row)
        if len(batch) >= size:
            yield batch
            batch = []
    if batch:
        yield batch

def process_users(csv_file: str) -> None:
    """Compose pipeline"""
    pipeline = batch_rows(
        enrich_data(
            filter_active(
                read_csv_rows(csv_file)
            )
        ),
        size=50
    )

    for batch_num, batch in enumerate(pipeline, 1):
        print(f"Batch {batch_num}: processing {len(batch)} users")
        for user in batch:
            save_to_db(user)

# Handles any CSV size with O(batch_size) memory
process_users("million_users.csv")

Expert Tips​

Tip 1: Delegate to sub-generators with yield from

def chain(*iterables):
    for it in iterables:
        yield from it   # Delegate all values from each iterable

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

Tip 2: Generator return value (return + StopIteration.value)

def gen_with_return():
    yield 1
    yield 2
    return "done"   # Passed as StopIteration.value

g = gen_with_return()
next(g)  # 1
next(g)  # 2
try:
    next(g)
except StopIteration as e:
    print(e.value)  # "done"

Tip 3: Check for generator functions with inspect.isgeneratorfunction()

import inspect

def normal_func(): return 42
def gen_func(): yield 42

print(inspect.isgeneratorfunction(normal_func))  # False
print(inspect.isgeneratorfunction(gen_func))     # True
Advertisement
Advertisement