Python生成器中yield from的强大功能

[QPY课程团队]

Python 生成器是编写高效、可读且内存友好代码时最好的工具之一。但在 Python 的生成器系统中，有一个隐藏的宝藏使得事情变得更加顺畅：yield from 语句。

在这篇文章中，我们将讨论：

1. yield from 是什么以及它是如何工作的。
2. 如何使用它构建干净、模块化的生成器工具。
3. 一个实际的例子：使用 FastAPI + yield from 从 Python 后端实时流式传输日志。

让我们开始吧！

生成器的快速回顾

生成器是一个函数，它允许你使用 yield 关键字一次返回一个值的序列。

示例:

def count_up_to(n):
    i = 1
    while i <= n:
        yield i
        i += 1

这不会返回一个列表，而是返回一个生成器对象，你可以对其进行循环遍历：

for number in count_up_to(3):
    print(number)

输出：

1 2 3

这种惰性求值在处理大文件、数据流或管道时非常有效，您只在需要时加载所需内容。

进入yield from

现在假设您想将部分生成器逻辑委托给另一个生成器。您可以这样做：

def wrapper():
    for value in count_up_to(3):
        yield value

但是，Python 提供了一种更简洁、更智能的方法：

def wrapper():
    yield from count_up_to(3)

只需一行代码，它的工作方式完全相同。 yield from 基本上将另一个可迭代对象或生成器中的所有值转发到当前的可迭代对象中。

示例 1：扁平化嵌套列表

让我们扁平化这个列表：

data = [1, [2, 3], [4, [5, 6]], 7]

我们想要： 1, 2, 3, 4, 5, 6, 7

实现方法如下：

def flatten(items):
    for item in items:
        if isinstance(item, list):
            yield from flatten(item)
        else:

yield item

用法：

for val in flatten([1, [2, 3], [4, [5, 6]], 7]):

print(val)

yield from 使得递归生成器变得轻而易举！

示例 2：组合多个数据源

想象一下，你正在从不同的数据源提取数据：

def fruits():

yield from ["苹果", "香蕉"]

def 蔬菜():
    yield from ["胡萝卜", "白萝卜"]

def 所有():
    yield from 水果()
    yield from 蔬菜()

everything()

将返回两个函数中的所有项，无需任何手动循环。这使得你的代码非常模块化。

---

实际用例：使用 yield from 和 FastAPI 流式传输日志

现在让我们将所学的知识应用于一个真实的后端用例：通过 HTTP 端点提供实时日志。

为什么使用生成器进行日志流式传输？

在后端中，你经常需要将日志流式传输到前端。这可能是为了：

• 实时调试
• 监控
• 管理仪表板

但是日志文件可能非常庞大，一次性加载它们是个坏主意。

这就是生成器（和 yield from）派上用场的地方：

• 逐行流式传输日志
• 几乎不占用内存
• 实时响应更新

步骤指南：构建实时日志流API

1. 安装 FastAPI 和 Uvicorn

pip install fastapi uvicorn

2. 创建流应用

# main.py
from fastapi import FastAPI

from fastapi.responses import StreamingResponse
import time
import os

app = FastAPI()

def tail_log_file(filepath):
    """一个生成器，从日志文件中逐行输出新内容，类似于 `tail -f`"""
    with open(filepath, "r") as f:
        # 移动到文件末尾
        f.seek(0, os.SEEK_END)
        while True:

line = f.readline()
if not line:
    time.sleep(0.5)  # 等待新行
    continue
yield line

def log_streamer():
    """一个可以使用 `yield from` 组合多个来源的包装器"""
    # 将来，您可以在这里使用来自其他来源的 yield
    yield from tail_log_file("app.log")

3. 添加 FastAPI 端点

@app.get("/logs")
def stream_logs():
    return StreamingResponse(log_streamer(), media_type="text/plain")

4. 运行应用

uvicorn main:app --reload

 

然后访问 http://localhost:8000/logs，实时查看日志流！

 

为什么 yield from 在这里很重要

你今天可能只从一个源进行流式传输，但明天你可能想从以下来源流式传输日志：

• 一个文件
• 一个子进程（例如 subprocess.Popen）
• 一个队列（例如 Kafka、Redis）
• 另一个 API

使用 yield from，你可以在不破坏结构的情况下组合和委托流。

示例：

def log_streamer():
    yield from tail_log_file("app.log")
    yield from tail_log_file("error.log")

 

这种模块化设计使得将来扩展您的应用程序变得简单。

最终思考

Python 的 yield from 不仅仅是一个语法快捷方式——它是一个强大的工具，用于编写干净、模块化和内存高效的生成器。

我们看到它如何帮助简化：

• 递归工具（如列表扁平化）
• 组合多个生成器
• 在生产就绪的后端进行实时流处理

借助像 FastAPI 这样的框架，您可以用几行代码构建优雅、高性能的数据流端点。

[作者]

帖子