pythonadvanced

Advanced Generators — send, yield from, close

Use generator.send(), yield from delegation, and close() for coroutine-style generators.

python
from typing import Generator

def running_average() -> Generator[float, float, None]:
    total = 0.0
    count = 0
    average = 0.0
    while True:
        value = yield average
        total += value
        count += 1
        average = total / count

avg = running_average()
next(avg)  # Prime the generator
print(avg.send(10))  # 10.0
print(avg.send(20))  # 15.0
print(avg.send(30))  # 20.0

# yield from — flatten nested lists
def flatten(nested: list) -> Generator:
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)
        else:
            yield item

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

# Pipeline of generators
def read_lines(path: str) -> Generator[str, None, None]:
    with open(path) as f:
        yield from (line.strip() for line in f)

def filter_non_empty(lines) -> Generator[str, None, None]:
    yield from (line for line in lines if line)

def to_upper(lines) -> Generator[str, None, None]:
    yield from (line.upper() for line in lines)

# Generator with cleanup
def managed_resource() -> Generator[str, None, None]:
    print("Acquiring resource")
    try:
        while True:
            yield "data"
    except GeneratorExit:
        print("Releasing resource")

gen = managed_resource()
next(gen)    # Acquiring resource
gen.close()  # Releasing resource

Use Cases

  • streaming pipelines
  • running statistics
  • generator delegation

Tags

Related Snippets

Similar patterns you can reuse in the same workflow.