yield from and Delegating Generators
yield from was introduced in Python 3.3 via PEP 380. It allows a generator to delegate work to another iterable or sub-generator. It is a powerful feature that forms the foundation of coroutines.
yield from Basic Behavior
yield from iterable passes all values from the iterable through the current generator as-is.
# Passing sub-iterable values without yield from
def chain_without_yield_from(*iterables):
for it in iterables:
for value in it: # Nested loop
yield value
# Concise with yield from
def chain_with_yield_from(*iterables):
for it in iterables:
yield from it # Delegation!
result = list(chain_with_yield_from([1, 2], [3, 4], [5, 6]))
print(result) # [1, 2, 3, 4, 5, 6]
# Applicable to all iterables: strings, lists, etc.
def flatten(nested):
for item in nested:
if isinstance(item, (list, tuple)):
yield from flatten(item) # Recursive delegation
else:
yield item
data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data))) # [1, 2, 3, 4, 5, 6, 7]
Value Passing and Exception Propagation
The real power of yield from goes beyond simple value passing — it provides a bidirectional channel.
# Delegating generator: the generator that uses yield from
# Subgenerator: the generator pointed to by yield from
def subgenerator():
"""Subgenerator"""
received = yield "first"
print(f"Subgenerator received: {received}")
received = yield "second"
print(f"Subgenerator received: {received}")
return "subgenerator done" # return value is passed to delegator
def delegating_gen():
"""Delegating generator"""
result = yield from subgenerator() # Receives the subgenerator's return value
print(f"Subgenerator result: {result}")
yield "after delegation"
gen = delegating_gen()
print(next(gen)) # "first"
print(gen.send("A")) # Subgenerator received: A → "second"
print(gen.send("B")) # Subgenerator received: B → subgenerator done → "after delegation"
What yield from handles:
- Forwards
next()to the subgenerator - Passes
send(value)values to the subgenerator - Passes
throw(exc)exceptions to the subgenerator - Returns the subgenerator's
StopIteration.valueas the result of theyield fromexpression
Recursive Generator Patterns
from typing import Any, Iterator
# Flatten nested structure
def deep_flatten(nested: Any) -> Iterator:
"""Flatten a nested structure of arbitrary depth"""
if isinstance(nested, (list, tuple, set)):
for item in nested:
yield from deep_flatten(item)
else:
yield nested
data = [1, [2, [3, [4, [5]]]]]
print(list(deep_flatten(data))) # [1, 2, 3, 4, 5]
# Tree structure traversal
from dataclasses import dataclass, field
@dataclass
class TreeNode:
value: int
children: list["TreeNode"] = field(default_factory=list)
def preorder(node: TreeNode) -> Iterator[int]:
"""Pre-order traversal (root → left → right)"""
yield node.value
for child in node.children:
yield from preorder(child) # Recursive delegation
def postorder(node: TreeNode) -> Iterator[int]:
"""Post-order traversal (left → right → root)"""
for child in node.children:
yield from postorder(child)
yield node.value
# Build tree
root = TreeNode(1, [
TreeNode(2, [TreeNode(4), TreeNode(5)]),
TreeNode(3, [TreeNode(6)])
])
print(list(preorder(root))) # [1, 2, 4, 5, 3, 6]
print(list(postorder(root))) # [4, 5, 2, 6, 3, 1]
Relationship with Coroutines
yield from was the underlying mechanism for implementing coroutines before asyncio. Python 3.5+'s async/await is built on top of yield from.
# Python 3.4 style coroutine (for historical understanding — use async/await in practice)
import asyncio
def old_style_coroutine():
"""yield from-based coroutine (for historical understanding)"""
result = yield from asyncio.sleep(1) # Communicates with asyncio
return result
# Modern style (actual usage)
async def modern_coroutine():
result = await asyncio.sleep(1) # async/await = syntactic sugar
return result
# Verifying exception propagation with yield from
def sub_gen():
try:
yield 1
yield 2
except RuntimeError as e:
print(f"Caught in sub: {e}")
yield "recovered after error"
def delegator():
yield from sub_gen()
gen = delegator()
print(next(gen)) # 1
gen.throw(RuntimeError, "external error injection") # Caught in sub: external error injection
# "recovered after error" output, then StopIteration
Practical: Tree Traversal, Pipelines
File System Traversal
from pathlib import Path
from typing import Iterator
def iter_files(directory: str | Path, pattern: str = "*") -> Iterator[Path]:
"""Recursively traverse all files in a directory"""
root = Path(directory)
for entry in root.iterdir():
if entry.is_dir():
yield from iter_files(entry, pattern) # Delegate to subdirectory
elif entry.match(pattern):
yield entry
# Find Python files
for py_file in iter_files("project/", "*.py"):
print(py_file)
Composing Generator Pipelines
from typing import Iterator, Callable, TypeVar
T = TypeVar("T")
def pipeline(*generators: Callable) -> Callable:
"""Utility that chains multiple generators into a pipeline"""
def apply(source):
result = source
for gen in generators:
result = gen(result)
return result
return apply
def read_lines(filepath: str) -> Iterator[str]:
with open(filepath, encoding="utf-8") as f:
yield from f
def strip_lines(lines: Iterator[str]) -> Iterator[str]:
for line in lines:
stripped = line.strip()
if stripped:
yield stripped
def parse_csv_line(lines: Iterator[str]) -> Iterator[list[str]]:
import csv
reader = csv.reader(lines)
yield from reader
def filter_header(rows: Iterator[list[str]]) -> Iterator[list[str]]:
first = True
for row in rows:
if first:
first = False
continue
yield row
# Compose pipeline
process = pipeline(strip_lines, parse_csv_line, filter_header)
data = process(read_lines("data.csv"))
for row in data:
print(row)
Recursive JSON Traversal
from typing import Any, Iterator
def iter_values(data: Any, path: str = "") -> Iterator[tuple[str, Any]]:
"""Traverse all leaf nodes of a nested JSON structure as (path, value) tuples"""
if isinstance(data, dict):
for key, value in data.items():
new_path = f"{path}.{key}" if path else key
yield from iter_values(value, new_path)
elif isinstance(data, list):
for i, item in enumerate(data):
yield from iter_values(item, f"{path}[{i}]")
else:
yield path, data
config = {
"database": {
"host": "localhost",
"port": 5432,
"credentials": {
"user": "admin",
"password": "secret"
}
},
"features": ["auth", "billing", "analytics"]
}
for path, value in iter_values(config):
print(f"{path}: {value}")
# database.host: localhost
# database.port: 5432
# database.credentials.user: admin
# database.credentials.password: secret
# features[0]: auth
# features[1]: billing
# features[2]: analytics
Expert Tips
Tip 1: yield from range(n) is faster than for x in range(n): yield x
# Slow way
def slow_gen(n):
for x in range(n):
yield x
# Fast way
def fast_gen(n):
yield from range(n)
Tip 2: Using the sub-generator's return value
def producer():
yield 1
yield 2
return {"count": 2, "status": "done"}
def consumer():
stats = yield from producer()
print(f"Stats: {stats}")
yield "done"
list(consumer())
# Stats: {'count': 2, 'status': 'done'}
Tip 3: Building generator chains with yield from
import itertools
def merged(*generators):
"""Connect multiple generators in sequence"""
for gen in generators:
yield from gen
# Same as itertools.chain but can add custom logic
result = list(merged(range(3), range(5, 8)))
print(result) # [0, 1, 2, 5, 6, 7]