FastAPI 流式传输数据实战用 StreamingResponse 与 yield 逐块下发字符串、二进制与大文件【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本文聚焦 FastAPI 从0.134.0起提供的原生流式传输能力在路径操作函数path operation function中直接声明response_classStreamingResponse并通过yield把纯字符串、原始bytes、超大文件乃至音视频数据**逐块chunk**发送给客户端全程不做 JSON 序列化、不把整份内容一次性加载进内存。读完本文你将掌握最省内存的大文件下载方案、适合对接AI LLM 输出的文本流式接口写法以及如何自定义StreamingResponse子类例如PNGStreamingResponse来声明正确的Content-Type。本文内容以官方法语文档 stream-data.md 为骨架并结合本仓库 docs_src/stream_data 下的可运行示例、路由层源码 与 测试用例 进行纵深解读。适用场景与边界什么时候该用 StreamingResponse文档开篇就划定了本主题的适用范围如果你要流式发送的数据本身可以结构化表达为 JSON例如每行一个 JSON 对象、需要被程序逐行解析应当优先使用专门的JSON Lines 流式传输方案参见官方指南 Diffuser des JSON Lines法文版。但如果你要的是纯二进制数据或纯字符串就用本主题介绍的StreamingResponse方案。典型的落地场景包括流式输出 LLM 生成的文本直接从 AI 大模型的输出流里逐段转发字符串客户端可以边生成边看到。流式传输大二进制文件按块读取、按块发送全程不需要把整个文件读入内存内存占用与文件大小无关。流式音视频边处理边生成边发送甚至可以让视频/音频在服务端实时编码后就立即推给客户端。从仓库测试可看出这是官方推荐路径test_stream_data/test_tutorial001.py 与 test_stream_data/test_tutorial002.py 分别对文本流与图片流做了端到端断言可以作为行为基准。版本提示文档特别注明此能力在FastAPI 0.134.0中引入原文 “Ajouté dans FastAPI 0.134.0”。这意味着路径操作函数可以直接作为生成器与response_classStreamingResponse搭配使用此前更常见的写法是在函数内部手工return StreamingResponse(...)。基础用法response_classStreamingResponseyield只要在路径操作函数的装饰器参数里声明response_classStreamingResponse函数本身就可以写成生成器用yield把每个数据块依次交给响应。完整的官方示例位于 docs_src/stream_data/tutorial001_py310.py核心骨架如下from collections.abc import AsyncIterable, Iterable from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() # 一段很长的文本按行切分成多个块 message ...一段较长的文本内容... app.get(/story/stream, response_classStreamingResponse) async def stream_story() - AsyncIterable[str]: for line in message.splitlines(): yield line要点在于FastAPI 会把yield出来的每个块原样交给StreamingResponse不会试图把它转成 JSON、也不会做任何 Pydantic 校验或二次序列化。这正是 StreamingResponse 与默认的JSONResponse行为上的根本差异。在 fastapi/routing.py 中可以看到 FastAPI 对生成器型路径操作函数的专门处理分支注释直接写明了 “Raw streaming with explicit response_class (e.g. StreamingResponse)”elif _is_async_gen_callable(dependant.call) or _is_gen_callable(dependant.call): # Raw streaming with explicit response_class (e.g. StreamingResponse) gen dependant.call(**solved_result.values) if _is_async_gen_callable(dependant.call): async def _async_stream_raw(async_gen): async for chunk in async_gen: yield chunk # 允许取消被触发 await anyio.sleep(0) gen _async_stream_raw(gen) response_args _build_response_args(...) response actual_response_class(contentgen, **response_args)即FastAPI 识别出函数是 async/sync 生成器后会跳过所有 JSON 序列化与响应模型过滤逻辑直接把生成器对象作为content喂给response_class也就是StreamingResponse。测试 test_tutorial001.py 也验证了返回的response.text与按行切分的原文完全一致证明各块内容被原样拼装下发。同步版本普通def函数同样可以流式传输并不要求必须用async def。文档明确说明可以用不带async的普通def函数同样写yieldapp.get(/story/stream-no-async, response_classStreamingResponse) def stream_story_no_async() - Iterable[str]: for line in message.splitlines(): yield line在docs_src/stream_data/tutorial001_py310.py中可以看到完整文件其实同时提供 async 与 sync 两种形态第 20–29 行两者行为完全等价供读者按自身代码风格取舍。可以省略返回类型注解流式传输不一定需要声明返回类型注解。因为 FastAPI 根本不会用 Pydantic 去把内容转成 JSON 或做序列化这种情况下注解的唯一作用就是给编辑器和静态检查工具看FastAPI 运行时不会读取它app.get(/story/stream-no-annotation, response_classStreamingResponse) async def stream_story_no_annotation(): for line in message.splitlines(): yield line这一点也意味着使用StreamingResponse时编码字节流的具体方式和责任完全在你手里——你自由决定每个块如何产生、如何编码注解不会约束你的输出。流式发送 bytes原始二进制字符串之外最重要的使用场景就是直接流式发送bytes。做法无非是对每个文本块手动做编码例如line.encode(utf-8)app.get(/story/stream-bytes, response_classStreamingResponse) async def stream_story_bytes() - AsyncIterable[bytes]: for line in message.splitlines(): yield line.encode(utf-8)在 tutorial001_py310.py 中文本流str块与字节流bytes块各提供四组变体——async def/def、有注解 / 无注解的交叉组合方便你验证各种写法均能正常工作。对字节流的测试同样断言了最终response.text与期望文本一致因为字节块最终会被拼接为同一份 UTF-8 内容。进阶自定义PNGStreamingResponse子类声明 Content-Type上面几个例子虽然成功把数据字节流式发出去了但响应没有Content-Type响应头客户端无法得知自己在接收什么类型的数据。解决方案是继承StreamingResponse在子类中通过类属性media_type声明内容类型。官方示例 docs_src/stream_data/tutorial002_py310.py 里定义了一个将Content-Type置为image/png的专用响应类class PNGStreamingResponse(StreamingResponse): media_type image/png然后在路径操作函数中用response_classPNGStreamingResponse启用它app.get(/image/stream, response_classPNGStreamingResponse) async def stream_image() - AsyncIterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunkmedia_type会被 Starlette/FastAPI 底层用于构造Content-Type: image/png头。这一点被测试明确锁定test_tutorial002.py 对所有图片流端点断言了response.headers[content-type] image/png且response.content mod.binary_image即发送的字节与源图片完全一致。同时该测试的 OpenAPI 快照显示声明了媒体类型后/openapi.json的 200 响应里会出现content: {image/png: {schema: {type: string}}}而 tutorial001 中未设置媒体类型的文本流端点其 OpenAPI 200 响应只有描述没有 content——这正是不做 JSON 序列化、不推断 schema的又一佐证。用 io.BytesIO 模拟文件为了把示例收敛在单文件内可直接复制运行官方示例用io.BytesIO在内存里伪造了一个文件对象。它只存活于内存但接口与真实文件一致可以像读文件那样迭代消费import base64 from io import BytesIO # 一个 Base64 编码的小 PNG 图片常量完整值见源码文件 image_base64 iVBORw0KGgo...完整常量在 tutorial002_py310.py 中 binary_image base64.b64decode(image_base64) def read_image() - BytesIO: return BytesIO(binary_image)文档的技术细节注记指出image_base64与binary_image这两个变量只是把图片先 Base64 编码、再解码成 bytes、最后塞进io.BytesIO纯粹是为了让例子可以自包含地复制运行完整且可直接运行的常量请参见仓库文件 tutorial002_py310.py。用 with 块保证文件对象被关闭上面stream_image里借助with上下文管理器包裹文件读取确保生成器函数含yield的函数执行完毕后、也就是整个响应发送完成之后文件对象被自动关闭。文档特别提醒本例因为是内存假文件io.BytesIO关闭与否并不那么关键但换成真实文件时务必保证工作结束后文件被正确关闭避免句柄泄漏。文件对象与 async避免阻塞事件循环文档强调大多数文件类对象默认并不兼容 async/await——例如它们没有await file.read()也不支持async for chunk in file而且很多情况下它们的读取是阻塞操作从磁盘或网络读取时可能卡住事件循环。不过上面的例子其实是个例外io.BytesIO已经整个存在于内存中读它不会阻塞任何东西。更普遍的情况是读真实文件/真实流会阻塞。规避方法把路径操作函数声明成普通def而不是async def这样 FastAPI 会把它放进线程池 workerthreadpool中执行避免阻塞主事件循环。仓库示例就提供了对应的同步版本app.get(/image/stream-no-async, response_classPNGStreamingResponse) def stream_image_no_async() - Iterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunk从实现细节看在 routing.py 中同步生成器函数会与 async 生成器走同一条 raw streaming 分支而内容进入StreamingResponse后底层Starlette 的StreamingResponse.__init__会对同步的可迭代内容用iterate_in_threadpool包装再以async for消费每个块因此同步生成器逐块迭代时的阻塞读取不会冻结事件循环。补充技巧来自原文档如果你需要在 async 函数内部调用阻塞代码、或在阻塞函数内调用 async 函数可以参考 FastAPI 同门的Asyncer库来解决跨同步/异步边界的问题。用yield from简化逐块转发当你迭代某个对象如文件类对象并对每个元素都执行yield时可以直接用yield from一次性透传每个元素、省掉for循环。这并非 FastAPI 专有语法而是纯 Python 特性但非常实用app.get(/image/stream-no-async-yield-from, response_classPNGStreamingResponse) def stream_image_no_async_yield_from() - Iterable[bytes]: with read_image() as image_file: yield from image_file注意yield from只适用于同步生成器透传把image_file的每次迭代结果逐个交给PNGStreamingResponseasync 场景下若需要透传异步迭代器则应使用async for显式转发。上面这个端点在 test_tutorial002.py 中同样被验证能正确返回image/png且内容完整。结合源码的完整工作流梳理把上述内容串起来一次典型的流式下发二进制文件实现包含四步定义文件来源真实打开文件用with open(path, rb)或内存模拟io.BytesIO。按块读取以固定缓冲迭代如for chunk in file_obj或显式file_obj.read(buffer_size)逐块产出避免整文件载入内存。自定义响应类如需正确媒体类型class XxxStreamingResponse(StreamingResponse): media_type application/octet-stream之类通过子类media_type属性声明Content-Type。在路径操作函数中组合声明response_classXxxStreamingResponse用async def或普通defyield逐块产出涉及真实磁盘/网络阻塞读取时优先用普通def让 FastAPI 在线程池中运行。行为基准可以参考仓库测试 test_tutorial002.py它同时覆盖了async 版、同步版、yield from版、无注解版以及无注解同步版共五个端点的状态码、Content-Type与字节完整度。小结当数据可以结构化为 JSON时优先走 JSON Lines 流式方案当数据是纯字符串或原始二进制时使用本文的StreamingResponseyield路线。FastAPI 会原样透传每个 yield 出来的块见 routing.py 的 raw streaming 分支不解析、不序列化返回类型注解只服务编辑器不影响运行时。需要正确Content-Type时继承StreamingResponse并设置类属性media_type如image/png。真实文件读取属于阻塞操作普通def生成的同步生成器会由 FastAPI/Starlette 放入线程池消费避免拖垮事件循环务必用with确保文件在使用后关闭也可以用yield from简化逐块转发。上述所有行为均有仓库测试背书文本流见 test_tutorial001.py图片/二进制流见 test_tutorial002.py完整可运行示例见 docs_src/stream_data/tutorial001_py310.py 与 docs_src/stream_data/tutorial002_py310.py。依赖说明StreamingResponse本身由 FastAPI 从 Starlette 直接再导出见 fastapi/responses.py因此上述用法在安装 FastAPI 时即可使用无需额外引入其他包本能力要求 FastAPI ≥ 0.134.0。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考