stream_with_context必须配合生成器函数使用，用于延续请求上下文至整个流式过程；直接在路由中yield会因上下文提前销毁而报RuntimeError。
Flask stream_with_context 必须配合生成器函数使用
直接在路由里写 yield 会报 RuntimeError: Response body closed，因为 Flask 默认响应对象生命周期短，生成器还没跑完，响应上下文就销毁了。必须用 stream_with_context 包裹生成器，把当前请求上下文延续到整个流式过程。
常见错误是把数据库查询和 yield 混在同一个函数里，但没处理连接生命周期——比如用 SQLAlchemy 时，session 在生成器外就 close 了，yield 到第二条数据就崩。
- 生成器函数要独立定义，不依赖局部 session 或 connection 对象
- 在生成器内部按需创建数据库连接（如用
create_engine(..., pool_pre_ping=True)），或传入已配置好的连接池实例 - 用
@stream_with_context装饰生成器调用，不是装饰生成器本身
from flask import Response, stream_with_context
<p>def data_generator():</p><h1>每次 yield 前建立轻量连接，查一批（如 1000 行），fetchmany 后 close</h1><pre class="brush:php;toolbar:false;">with engine.connect() as conn:
result = conn.execute(text("SELECT * FROM huge_table LIMIT 100000"))
while True:
rows = result.fetchmany(1000)
if not rows:
break
for row in rows:
yield f"{row.id},{row.name}\n".encode("utf-8")
@app.route('/export') def export(): return Response(stream_with_context(data_generator()), mimetype='text/csv')
避免 jsonify 和模板渲染干扰流式输出
jsonify 会强制把整个返回体转成 JSON 字符串并加 Content-Type: application/json，彻底破坏流式结构；用 render_template 渲染 HTML 更不行——它等所有数据收集完才输出，根本不是流。
导出场景下，你真正需要的是原始字节流：CSV、TSV、JSON Lines（每行一个 JSON 对象）、甚至自定义二进制分块。这些都得绕过 Flask 的高层封装，直操作 Response 构造。
- 禁用
jsonify，改用yield json.dumps(item).encode("utf-8") + b"\n" - 不要在生成器里调
url_for、get_flashed_messages等依赖完整请求上下文的函数 - 如果要用 Jinja 渲染单行内容（比如 CSV 头），用
Template("{{ a }},{{ b }}").render(a=row.a, b=row.b)，别碰全局render_template
超时与客户端断连必须主动检测
导出可能持续几分钟，Nginx 默认 60 秒超时，Gunicorn worker timeout 默认 30 秒，浏览器也可能主动断开。Flask 不会自动感知客户端是否还在线，继续 yield 就是白耗资源。
关键是在每次 yield 前检查 socket 是否可用。虽然 Python 没有跨平台的“客户端是否存活”API，但可以利用底层 WSGI environ['wsgi.file_wrapper'] 或更可靠的方式：捕获 BrokenPipeError 和 ConnectionResetError，并在生成器中做防御性处理。
- 用
try/except (BrokenPipeError, ConnectionResetError)包裹 yield 语句 - 捕获后立即
return，终止生成器，避免后续查询和计算 - Nginx 配置要同步调大：
proxy_read_timeout 600;、proxy_send_timeout 600;
def safe_yield(generator):
for chunk in generator:
try:
yield chunk
except (BrokenPipeError, ConnectionResetError):
return # 客户端已断，停止推送
<p>@app.route('/export')
def export():
return Response(stream_with_context(safe_yield(data_generator())),
mimetype='text/csv')
</p>
内存不增长 ≠ 不卡顿：注意生成器里的 I/O 阻塞点
很多人以为用了生成器就“不占内存”，结果导出卡在 20% 不动——问题往往不在 Python 层，而在数据库驱动或网络层阻塞。比如用 psycopg2 默认配置查千万级表，fetchmany(1000) 可能实际拉回全部结果再切片；MySQLdb 更甚，不设 cursorclass=DictCursor 加 sscursor（server-side cursor）就会全量加载。
- PostgreSQL：用
conn.cursor(name='large_export')创建命名游标，再fetchmany() - MySQL：用
pymysql.cursors.SSCursor或mysqlclient的sscursor=True - 避免在生成器里做复杂字符串拼接或 JSON 序列化——提前编译正则、复用
json.JSONEncoder实例
真正压测时，瓶颈常在磁盘 I/O 或网卡带宽，而不是 Python 解释器。这时候 yield 间隔哪怕只有几毫秒，对用户体验影响也远小于一次 500ms 的慢查询。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
