Python yield from 语法

yield from 用于将迭代任务委托给另一个生成器，简化嵌套迭代代码。

基本用法

对比传统写法

# 传统方式：手动迭代
def chain_manual(*iterables):
    for iterable in iterables:
        for item in iterable:
            yield item


# yield from 方式：自动委托
def chain_from(*iterables):
    for iterable in iterables:
        yield from iterable


list(chain_manual([1, 2], [3, 4]))    # [1, 2, 3, 4]
list(chain_from([1, 2], [3, 4]))      # [1, 2, 3, 4]

扁平化嵌套结构

def flatten(nested):
    "递归展平嵌套列表"
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)  # 委托给递归调用
        else:
            yield item


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

委托子生成器

yield from 建立委托生成器与子生成器的双向通道：

def accumulator():
    "累加器子生成器"
    total = 0
    while True:
        value = yield total
        if value is None:
            break
        total += value
    return total  # 返回最终值


def aggregator():
    "委托生成器"
    result = yield from accumulator()  # 接收子生成器返回值
    print(f"最终结果: {result}")
    return result


gen = aggregator()
next(gen)           # 启动，返回 0
gen.send(10)        # 返回 10
gen.send(20)        # 返回 30
gen.send(None)      # 输出: 最终结果: 30，抛出 StopIteration(30)

双向通信

yield from 自动转发 send、throw 和 close 调用：

def sub_gen():
    "子生成器"
    received = yield "ready"
    yield f"got: {received}"
    return "done"


def main_gen():
    "主生成器"
    result = yield from sub_gen()
    return f"sub returned: {result}"


gen = main_gen()
print(next(gen))       # ready
print(gen.send("hello"))  # got: hello
try:
    gen.send(None)
except StopIteration as e:
    print(e.value)     # sub returned: done

处理异常

def sub_gen():
    try:
        yield 1
        yield 2
        yield 3
    except ValueError as e:
        yield f"caught: {e}"


def main_gen():
    yield from sub_gen()


gen = main_gen()
print(next(gen))  # 1
print(gen.throw(ValueError, "error"))  # caught: error
print(next(gen))  # 3

协程协作示例

async def task_a():
    yield from range(3)
    return "A done"


async def task_b():
    yield from range(3, 6)
    return "B done"


def run_tasks():
    "顺序执行多个协程"
    result_a = yield from task_a()
    result_b = yield from task_b()
    return f"{result_a}, {result_b}"


gen = run_tasks()
list(gen)  # [0, 1, 2, 3, 4, 5]

实际应用

遍历文件树

import os


def walk_files(directory):
    "递归遍历目录"
    for name in os.listdir(directory):
        path = os.path.join(directory, name)
        if os.path.isdir(path):
            yield from walk_files(path)  # 委托给子目录
        else:
            yield path


# 使用
for file_path in walk_files("/project"):
    print(file_path)

组合多个数据源

def read_chunks(file, size=1024):
    "分块读取文件"
    while chunk := file.read(size):
        yield chunk


def process_files(*paths):
    "处理多个文件"
    for path in paths:
        with open(path, 'rb') as f:
            yield from read_chunks(f)

yield from 执行流程

yield from EXPR 执行过程：

1. 获取迭代器：iter(EXPR)
2. 建立双向通道
3. 循环迭代：
   - yield 每个值给调用者
   - 转发 send() 值给子生成器
   - 转发 throw() 异常给子生成器
   - 转发 close() 调用给子生成器
4. 子生成器结束后获取返回值
5. 继续执行后续代码

对比总结

要点总结

yield from 是生成器委托的语法糖，建立了主生成器与子生成器的透明通道。

D:\git2\jwdev\articles\PYTHON\进阶\迭代器与生成器\yield from语法.md