Generators & Lazy Evaluation¶

def count_up(start, stop):
    """Generator: yields integers from start to stop, inclusive."""
    current = start
    while current <= stop:
        print(f"  >> about to yield {current}")
        yield current
        print(f"  >> resumed after yield {current}")
        current += 1
    print("  >> generator exhausted")

gen = count_up(1, 3)
print(type(gen))           # <class 'generator'>
print("Calling next()...")
print(next(gen))           # >> about to yield 1  |  1
print(next(gen))           # >> resumed ...  |  >> about to yield 2  |  2
for val in gen:            # drives remaining values
    print(val)             # 3

# Second call creates a fresh generator – they don't restart
gen2 = count_up(1, 3)
print(list(gen2))   # [1, 2, 3]

import itertools

# Infinite generator – never exhausted
def integers(start=0):
    n = start
    while True:
        yield n
        n += 1

# Pipeline stages: each is a generator consuming another generator
def evens(src):
    for n in src:
        if n % 2 == 0:
            yield n

def squared(src):
    for n in src:
        yield n * n

def take(n, src):
    for _ in range(n):
        yield next(src)

# Compose the pipeline (nothing executes yet)
pipeline = take(5, squared(evens(integers(1))))

# Pull values through on demand
print(list(pipeline))   # [4, 16, 36, 64, 100]
#   integers(1)  → 1,2,3,4,5,6,7,8,9,10,...
#   evens        → 2,4,6,8,10,...
#   squared      → 4,16,36,64,100,...
#   take(5)      → [4,16,36,64,100]  (stops pulling)

def flatten(nested):
    """Recursively flatten arbitrarily nested lists."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)   # delegate to recursive call
        else:
            yield item

data = [1, [2, [3, 4], 5], [6, 7], 8]
print(list(flatten(data)))   # [1, 2, 3, 4, 5, 6, 7, 8]

# Without yield from, you'd need:
# for sub_item in flatten(item):
#     yield sub_item
# yield from is cleaner AND passes send() / throw() / close() through
# to the sub-generator (critical for coroutine chaining)

# yield from also captures the sub-generator's StopIteration value
def sub():
    yield 1
    yield 2
    return "sub_done"          # StopIteration value

def delegator():
    result = yield from sub()  # 'result' gets "sub_done"
    print(f"sub returned: {result}")
    yield 99

gen = delegator()
print(next(gen))   # 1
print(next(gen))   # 2
print(next(gen))   # sub returned: sub_done  |  99

def running_average():
    """Coroutine that accepts values via send() and yields the average."""
    total = 0.0
    count = 0
    value = yield   # prime: first next() runs to here, returns None
    while True:
        total += value
        count += 1
        value = yield total / count   # send average back, receive next value

avg = running_average()
next(avg)          # prime the coroutine (advance to first yield)
print(avg.send(10))    # 10.0
print(avg.send(20))    # 15.0
print(avg.send(30))    # 20.0
print(avg.send(40))    # 25.0

# throw() injects an exception at the yield point
def resilient():
    try:
        while True:
            value = yield
            print(f"Got: {value}")
    except ValueError as e:
        print(f"Caught inside generator: {e}")
        yield "handled"

r = resilient()
next(r)
r.send(42)               # Got: 42
print(r.throw(ValueError, "bad input"))   # Caught... | 'handled'

# close() raises GeneratorExit at the yield point
gen = (x for x in range(100))
print(next(gen))   # 0
gen.close()        # generator is now exhausted / closed

# This for loop:
for x in [1, 2, 3]:
    print(x)

# Is exactly equivalent to:
_iter = iter([1, 2, 3])    # calls list.__iter__()
while True:
    try:
        x = next(_iter)    # calls _iter.__next__()
        print(x)
    except StopIteration:
        break

# Generators satisfy the iterator protocol natively:
# gen.__iter__() returns self
# gen.__next__() resumes and returns the next yielded value
# So generators can be used directly in for loops, zip(), map() etc.

import sys, tracemalloc

N = 1_000_000

# Eager list
tracemalloc.start()
result_list = [n * n for n in range(N)]
_, list_peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

# Lazy generator
tracemalloc.start()
result_gen = (n * n for n in range(N))
_, gen_peak = tracemalloc.get_traced_memory()
tracemalloc.stop()

print(f"List peak:      {list_peak / 1_048_576:.2f} MB")
print(f"Generator peak: {gen_peak / 1_048_576:.4f} MB")
print(f"Ratio: {list_peak // gen_peak}x more memory for list")

import csv, gzip, itertools
from pathlib import Path
from typing import Iterator

def read_gzipped_csv(path: Path) -> Iterator[dict]:
    """Stream rows from a gzipped CSV without loading it into memory."""
    with gzip.open(path, "rt", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def parse_numerics(rows: Iterator[dict]) -> Iterator[dict]:
    """Cast revenue and quantity to correct types."""
    for row in rows:
        row["revenue"]  = float(row.get("revenue",  0) or 0)
        row["quantity"] = int(row.get("quantity", 0)   or 0)
        yield row

def filter_region(rows: Iterator[dict], region: str) -> Iterator[dict]:
    for row in rows:
        if row.get("region") == region:
            yield row

def aggregate_by_product(rows: Iterator[dict]) -> dict:
    totals = {}
    for row in rows:
        pid = row["product_id"]
        totals[pid] = totals.get(pid, 0.0) + row["revenue"]
    return totals

# Compose pipeline – nothing runs yet
files = Path("data").glob("sales_*.csv.gz")   # e.g., 50 files @ 1 GB each
raw      = itertools.chain.from_iterable(read_gzipped_csv(f) for f in files)
parsed   = parse_numerics(raw)
apac     = filter_region(parsed, "APAC")
result   = aggregate_by_product(apac)   # pulls the pipeline

# Peak memory stays near the size of one CSV row, never the full dataset
print(sorted(result.items(), key=lambda x: x[1], reverse=True)[:5])

from functools import wraps

def coroutine(fn):
    """Decorator to auto-prime a generator coroutine."""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        gen = fn(*args, **kwargs)
        next(gen)   # advance to first yield
        return gen
    return wrapper

@coroutine
def grep(pattern):
    """Coroutine that receives lines and prints matches."""
    import re
    while True:
        line = yield
        if re.search(pattern, line):
            print(f"  MATCH: {line.rstrip()}")

@coroutine
def broadcast(*targets):
    """Fan-out: send each received value to multiple coroutines."""
    while True:
        value = yield
        for target in targets:
            target.send(value)

# Build a coroutine pipeline
errors  = grep(r"ERROR")
timeouts = grep(r"timeout")
router   = broadcast(errors, timeouts)

lines = [
    "2024-01-01 INFO  service started\n",
    "2024-01-01 ERROR disk full\n",
    "2024-01-01 ERROR connection timeout\n",
    "2024-01-01 DEBUG heartbeat ok\n",
]

for line in lines:
    router.send(line)

router.close()