FastAPI 流式传输实战用 StreamingResponse 逐块发送字符串、字节与大文件0.134.0【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本指南聚焦 FastAPI 中的原始二进制数据 / 纯字符串流式传输方案当响应体不是「结构化 JSON」而是需要边生成边发送的文本如 AI LLM 输出、大文件、音视频数据时如何借助StreamingResponse与生成器函数逐块下发。读完本文将掌握yield与response_class的组合写法、同步与异步函数的取舍、自定义媒体类型响应类、io.BytesIO模拟文件读取以及yield from等实用技巧并能直接用 示例源码 与 仓库内测试 加以验证。版本提示本指南介绍的能力由官方文档标注为FastAPI 0.134.0 起引入见 英文原文 的版本注记请确认所用 FastAPI 不低于该版本。一、先分清两条路线结构化 JSON 数据走「JSON Lines」纯数据流才用本文方案在动手之前需要先明确一个概念边界什么时候该用本文介绍的方法什么时候不该用。官方文档西语版原文档在开篇就给出了判断标准如果你要传输可以按 JSON 结构化的数据例如逐行输出结构化记录应当优先参考 Transmitir JSON Lines流式 JSON Lines指南。仓库中也提供对应示例代码 docs_src/stream_json_lines。如果你要传输的是纯二进制数据或纯字符串——例如直接转发某个 AI LLM 服务的原始文本输出——才采用本文介绍的StreamingResponse原始流方案。之所以做这种区分关键在于 FastAPI 默认会对返回数据进行 JSON 序列化处理而本文要讨论的场景恰恰要求FastAPI 不做任何序列化、不尝试解析内容把每一块数据原样交给客户端。这也直接决定了下面代码中「不需要返回值注解」等一系列写法的合理性。二、典型使用场景结合文档说明流式传输原始数据主要覆盖三类实际需求AI / LLM 服务的原始文本输出LLM 通常以流式方式逐词或逐块吐出结果服务端拿到这些纯字符串后可以不经任何缓冲地即时转发给前端显著降低「首字延迟」。大体积二进制文件的流式读取逐块读取、逐块发送而不需要把整个文件一次性载入内存这对内存占用是本质性的优化。视频 / 音频流不仅可以流式转发已存在的媒体文件甚至可以做到「边处理边生成边发送」例如实时转码或合成后再把结果块陆续推出。三、核心写法response_classStreamingResponseyield只要在path operation function上声明response_classStreamingResponse就可以在函数体内用yield逐次产出每一块数据。官方教程的第一个示例完整代码如下对应仓库文件 docs_src/stream_data/tutorial001_py310.pyfrom collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() message Rick: (stumbles in drunkenly, and turns on the lights) Morty! You gotta come on. You got--... you gotta come with me. Morty: (rubs his eyes) What, Rick? Whats going on? Rick: I got a surprise for you, Morty. Morty: Its the middle of the night. What are you talking about? Rick: (spills alcohol on Mortys bed) Come on, I got a surprise for you. (drags Morty by the ankle) Come on, hurry up. (pulls Morty out of his bed and into the hall) Morty: Ow! Ow! Youre tugging me too hard! Rick: We gotta go, gotta get outta here, come on. Got a surprise for you Morty. app.get(/story/stream, response_classStreamingResponse) async def stream_story() - AsyncIterable[str]: for line in message.splitlines(): yield line这段代码的关键在于最后三行message.splitlines()把文本切成若干行yield把每一行单独作为一个数据块交出去。文档明确指出FastAPI 会把这些数据块原样交给StreamingResponse不会尝试把它们转换为 JSON也不会做任何类似处理。从实现层面看fastapi.responses模块中的StreamingResponse直接重导出自 Starlette 的同名类见 fastapi/responses.py#L8-L13也就是说流式传输的底层机制由成熟的starlette.responses.StreamingResponse提供FastAPI 侧的工作是识别生成器并保持其逐块语义不被 JSON 化拦截。3.1 使用普通def函数非 async同样可以你完全可以使用不带async的普通def函数配合yield效果一致。仓库示例 docs_src/stream_data/tutorial001_py310.py#L26-L29 给出了对照版本app.get(/story/stream-no-async, response_classStreamingResponse) def stream_story_no_async() - Iterable[str]: for line in message.splitlines(): yield line注意此时返回类型注解应写成同步可迭代类型Iterable[str]需要额外from collections.abc import Iterable。选择async def还是def的真正标准并非「想不想用异步语法」而是函数体内是否有阻塞式 IO——这一点在第 7 节会结合文件读取详细讨论。3.2 完全不需要返回值注解与普通的返回 JSON 的路由不同流式传输二进制数据时你并不需要声明返回类型注解。原因是 FastAPI 根本不会用 Pydantic 把这段数据转成 JSON也不会以任何方式序列化它此时类型注解仅仅服务于你的编辑器和静态检查工具FastAPI 不会读取它。对应示例见 docs_src/stream_data/tutorial001_py310.py#L32-L35app.get(/story/stream-no-annotation, response_classStreamingResponse) async def stream_story_no_annotation(): for line in message.splitlines(): yield line文档对此还有一句意味深长的总结既然 FastAPI 不介入序列化那么使用StreamingResponse就意味着你同时获得自由与责任——最终发出的每一个字节都由你自己精确地产生和编码与类型注解无关。这要求开发者对内容编码、块切分、字符集等细节心中有数。3.3 传输bytes把字符串编码后逐块 yield文档列举的首要场景之一就是发送bytes而不是str。方法与字符串完全一致只需在产出前把文本编码为字节例如 docs_src/stream_data/tutorial001_py310.py#L44-L47app.get(/story/stream-bytes, response_classStreamingResponse) async def stream_story_bytes() - AsyncIterable[bytes]: for line in message.splitlines(): yield line.encode(utf-8)AsyncIterable[bytes]这种注解对混合型数据流同样适用——你可以自由选择让yield交出字符串或字节FastAPI 不会帮你做隐式编码。示例源码文件还提供了上述两种变体在「async / 非 async」「有无注解」交叉组合下的完整版本共 8 个端点见 docs_src/stream_data/tutorial001_py310.py方便逐一对照实验。四、自定义PNGStreamingResponse用media_type子类补充Content-Type前文的示例虽然能正确流式输出数据字节但存在一个明显缺陷响应没有携带Content-Type头客户端根本不知道收到的数据是什么类型。解决办法是自定义StreamingResponse的子类通过类属性media_type指定响应头Content-Type。官方教程第二步的示例docs_src/stream_data/tutorial002_py310.py定义了一个图片流响应类class PNGStreamingResponse(StreamingResponse): media_type image/png然后在路由中把它作为response_class使用docs_src/stream_data/tutorial002_py310.py#L23-L27app.get(/image/stream, response_classPNGStreamingResponse) async def stream_image() - AsyncIterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunk这一设计有两个可见收益HTTP 层面客户端能从响应头正确识别数据类型如image/png浏览器也能直接渲染。OpenAPI 层面由于media_type已知FastAPI 生成的 OpenAPI schema 会把该接口的200响应媒体类型标记为image/png。这一点被仓库测试明确断言见下文第 8 节。4.1 用io.BytesIO在内存中模拟文件read_image()返回的是一个io.BytesIO对象——一种只存在于内存中的 file-like 对象但它提供了与真实文件一致的接口可以with打开、可以迭代逐块消费。对应定义见 docs_src/stream_data/tutorial002_py310.py#L12-L13def read_image() - BytesIO: return BytesIO(binary_image)示例中另外两个变量只是为了让演示代码能独立运行而准备的image_base64是一段 Base64 编码的小图片文本binary_image base64.b64decode(image_base64)把它解码成字节串再塞进io.BytesIO见 docs_src/stream_data/tutorial002_py310.py#L1-L13。你可以把整份文件原样复制下来直接运行无需任何外部资源。4.2 为什么用with块包裹 file-like 对象注意流式端点中with read_image() as image_file:的写法with块保证了当生成器函数结束即响应发送完毕后这个 file-like 对象会被关闭。在io.BytesIO这个具体例子里关闭与否影响不大——它是内存假文件但作者特意强调换成真实文件时确保用完后关闭文件非常重要。因此把「读取—逐块 yield」放进with上下文管理器中是处理真实文件流的推荐写法。五、文件读取与 async 的关系别让阻塞式读文件卡住 event loop大部分 file-like 对象默认并不支持 async / await它们既没有await file.read()也没有async for chunk in file这种接口。而且很多情况下读取它们属于阻塞式操作从磁盘或网络读取如果放在事件循环里直接执行会阻塞整个应用的并发处理能力。这里有个值得注意的反例上一节的io.BytesIO是例外因为数据已经全部在内存里读它不会阻塞任何东西——所以示例中即使写了async def也毫无问题。但如果面对的是真实的磁盘文件或网络流正确姿势是把path operation function声明为普通def。这样 FastAPI 会把它放进线程池threadpool的 worker 中执行yield的生成逻辑在独立线程里逐块产出从而避免阻塞主事件循环。官方示例见 docs_src/stream_data/tutorial002_py310.py#L30-L34app.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这本质上复用了 FastAPI 对同步路径操作函数的通用调度策略async def直接在事件循环运行普通def则交给线程池。当你面对「同步读取 流式输出」的组合时把整个路由声明为def是既简单又安全的选择。进阶提示如果你需要在 async 函数内部调用阻塞代码或在阻塞函数内部调用 async 代码文档推荐使用 FastAPI 的姊妹库 Asyncer 来桥接——这在混用同步 IO 与异步生成器的复杂链路中会很有用。六、用yield from跳过手写 for 循环当你正在迭代某个可迭代对象比如 file-like 对象并且打算把每个元素原样 yield 出去时Python 提供了一种简写yield from。它会替你逐个转发被迭代对象的元素从而省掉一层for循环。这不是 FastAPI 特有的语法而是纯粹的 Python 技巧但在流式响应中非常实用。官方示例 docs_src/stream_data/tutorial002_py310.py#L37-L40 演示了把整段for chunk in image_file: yield chunk压缩成一行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_fileimage_fileBytesIO本身是一个可迭代对象yield from image_file会逐个产出其中的字节块效果与逐块yield完全一致代码却更简洁。若生成器内还包含收尾逻辑例如这里的with关闭yield from也不会破坏它——文件依旧会在生成结束后被正确关闭。七、完整端点矩阵一览两个教程文件共同覆盖了「文本 / 图片文件」×「async / 非 async」×「有注解 / 无注解」×「for / yield from」的多种组合。这里汇总所有端点的路径与语义方便读者对照实验文本流示例tutorial001_py310.py路径函数形态数据形式/story/streamasync def注解AsyncIterable[str]字符串行/story/stream-no-asyncdef注解Iterable[str]字符串行/story/stream-no-annotationasync def无注解字符串行/story/stream-no-async-no-annotationdef无注解字符串行/story/stream-bytesasync def注解AsyncIterable[bytes]UTF-8 编码字节/story/stream-no-async-bytesdef注解Iterable[bytes]编码字节/story/stream-no-annotation-bytesasync def无注解编码字节/story/stream-no-async-no-annotation-bytesdef无注解编码字节图片流示例tutorial002_py310.py路径函数形态要点/image/streamasync defwithfor chunk/image/stream-no-asyncdef线程池中同步读取/image/stream-no-async-yield-fromdefyield from image_file/image/stream-no-annotationasync def无注解—/image/stream-no-async-no-annotationdef无注解—八、用仓库测试验证一切行为这两个教程文件并非孤立的代码片段仓库为它们配套了完整的自动化测试是理解行为边界的绝佳参考tests/test_tutorial/test_stream_data/test_tutorial001.py通过TestClient对/story/stream系列全部 8 个端点逐一断言每个路径都返回 HTTP200响应体文本与预期台词完全一致即多行台词被按行流式拼接后内容不丢失、不重排顺带对/openapi.json做快照断言确认各端点生成的 OpenAPI 描述符合预期没有多余的响应体 schema。tests/test_tutorial/test_stream_data/test_tutorial002.py对/image/stream系列 5 个端点断言响应状态码为200响应头content-type等于image/png——验证了自定义media_type子类确实生效响应字节内容与mod.binary_image逐字节一致——证明with、for chunk、yield from、有无注解等写法的产出完全相同同时断言 OpenAPI 中该接口的响应媒体类型被标记为image/png。这两个测试文件事实上回答了一个常见疑问「这么多等价写法结果到底一不一样」——答案是对客户端而言文本/字节内容与媒体类型完全一致不同写法主要影响的是实现风格与服务端的调度方式。九、启动与验证方式本仓库基于pyproject.toml与uv.lock管理依赖。若要在本地体验上述流式端点可参考仓库脚本 scripts/test.sh 与 scripts/lint.sh 了解测试与静态检查的执行约定并用如下常规方式手动验证以docs_src为可导入包把示例文件作为模块启动例如uvicorn docs_src.stream_data.tutorial001_py310:app --reload需在仓库根目录运行并确保 fastapi、uvicorn 等依赖已按 pyproject.toml 安装。用支持流式读取的客户端观察效果。对文本端点可用curl直观看到分块效果curl -N http://127.0.0.1:8000/story/stream-N即--no-buffer会让 curl 不做缓冲分块到达即打印便于体会「逐块下发」的过程。对图片端点验证响应头与字节完整性curl -N -D - http://127.0.0.1:8000/image/stream -o out.png观察输出中Content-Type: image/png并可用file out.png确认它确实是一张合法 PNG。若需回归验证全部端点行为直接运行仓库自带测试pytest tests/test_tutorial/test_stream_data/十、小结给流式端点写出正确代码的三个决策点综合文档与仓库源码为「原始数据流」编写 FastAPI 端点时只需要在三个决策点上做选择选方案结构化 JSON 数据 → 参考 JSON Lines 流式指南纯字符串 / 二进制 → 用本文的response_classStreamingResponse方案。定形态函数体内只有内存对象如io.BytesIO且无阻塞 IO可以用async def涉及磁盘、网络等阻塞读取用普通def交给线程池避免卡住事件循环。补语义用自定义子类的media_type让响应携带正确的Content-Type同时让 OpenAPI 文档中的媒体类型正确呈现用with块管理 file-like 对象的生命周期能用yield from的场景就少写一层循环。三个决策全部落在 tutorial001_py310.py 与 tutorial002_py310.py 两个示例文件里而正确性则由 test_tutorial001.py 与 test_tutorial002.py 钉死。按图索骥即可在真实项目中安全、高效地落地 AI 文本流、大文件与音视频流的服务端输出。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastAPI 流式传输实战：用 StreamingResponse 逐块发送字符串、字节与大文件（0.134.0+）
FastAPI 流式传输实战用 StreamingResponse 逐块发送字符串、字节与大文件0.134.0【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本指南聚焦 FastAPI 中的原始二进制数据 / 纯字符串流式传输方案当响应体不是「结构化 JSON」而是需要边生成边发送的文本如 AI LLM 输出、大文件、音视频数据时如何借助StreamingResponse与生成器函数逐块下发。读完本文将掌握yield与response_class的组合写法、同步与异步函数的取舍、自定义媒体类型响应类、io.BytesIO模拟文件读取以及yield from等实用技巧并能直接用 示例源码 与 仓库内测试 加以验证。版本提示本指南介绍的能力由官方文档标注为FastAPI 0.134.0 起引入见 英文原文 的版本注记请确认所用 FastAPI 不低于该版本。一、先分清两条路线结构化 JSON 数据走「JSON Lines」纯数据流才用本文方案在动手之前需要先明确一个概念边界什么时候该用本文介绍的方法什么时候不该用。官方文档西语版原文档在开篇就给出了判断标准如果你要传输可以按 JSON 结构化的数据例如逐行输出结构化记录应当优先参考 Transmitir JSON Lines流式 JSON Lines指南。仓库中也提供对应示例代码 docs_src/stream_json_lines。如果你要传输的是纯二进制数据或纯字符串——例如直接转发某个 AI LLM 服务的原始文本输出——才采用本文介绍的StreamingResponse原始流方案。之所以做这种区分关键在于 FastAPI 默认会对返回数据进行 JSON 序列化处理而本文要讨论的场景恰恰要求FastAPI 不做任何序列化、不尝试解析内容把每一块数据原样交给客户端。这也直接决定了下面代码中「不需要返回值注解」等一系列写法的合理性。二、典型使用场景结合文档说明流式传输原始数据主要覆盖三类实际需求AI / LLM 服务的原始文本输出LLM 通常以流式方式逐词或逐块吐出结果服务端拿到这些纯字符串后可以不经任何缓冲地即时转发给前端显著降低「首字延迟」。大体积二进制文件的流式读取逐块读取、逐块发送而不需要把整个文件一次性载入内存这对内存占用是本质性的优化。视频 / 音频流不仅可以流式转发已存在的媒体文件甚至可以做到「边处理边生成边发送」例如实时转码或合成后再把结果块陆续推出。三、核心写法response_classStreamingResponseyield只要在path operation function上声明response_classStreamingResponse就可以在函数体内用yield逐次产出每一块数据。官方教程的第一个示例完整代码如下对应仓库文件 docs_src/stream_data/tutorial001_py310.pyfrom collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() message Rick: (stumbles in drunkenly, and turns on the lights) Morty! You gotta come on. You got--... you gotta come with me. Morty: (rubs his eyes) What, Rick? Whats going on? Rick: I got a surprise for you, Morty. Morty: Its the middle of the night. What are you talking about? Rick: (spills alcohol on Mortys bed) Come on, I got a surprise for you. (drags Morty by the ankle) Come on, hurry up. (pulls Morty out of his bed and into the hall) Morty: Ow! Ow! Youre tugging me too hard! Rick: We gotta go, gotta get outta here, come on. Got a surprise for you Morty. app.get(/story/stream, response_classStreamingResponse) async def stream_story() - AsyncIterable[str]: for line in message.splitlines(): yield line这段代码的关键在于最后三行message.splitlines()把文本切成若干行yield把每一行单独作为一个数据块交出去。文档明确指出FastAPI 会把这些数据块原样交给StreamingResponse不会尝试把它们转换为 JSON也不会做任何类似处理。从实现层面看fastapi.responses模块中的StreamingResponse直接重导出自 Starlette 的同名类见 fastapi/responses.py#L8-L13也就是说流式传输的底层机制由成熟的starlette.responses.StreamingResponse提供FastAPI 侧的工作是识别生成器并保持其逐块语义不被 JSON 化拦截。3.1 使用普通def函数非 async同样可以你完全可以使用不带async的普通def函数配合yield效果一致。仓库示例 docs_src/stream_data/tutorial001_py310.py#L26-L29 给出了对照版本app.get(/story/stream-no-async, response_classStreamingResponse) def stream_story_no_async() - Iterable[str]: for line in message.splitlines(): yield line注意此时返回类型注解应写成同步可迭代类型Iterable[str]需要额外from collections.abc import Iterable。选择async def还是def的真正标准并非「想不想用异步语法」而是函数体内是否有阻塞式 IO——这一点在第 7 节会结合文件读取详细讨论。3.2 完全不需要返回值注解与普通的返回 JSON 的路由不同流式传输二进制数据时你并不需要声明返回类型注解。原因是 FastAPI 根本不会用 Pydantic 把这段数据转成 JSON也不会以任何方式序列化它此时类型注解仅仅服务于你的编辑器和静态检查工具FastAPI 不会读取它。对应示例见 docs_src/stream_data/tutorial001_py310.py#L32-L35app.get(/story/stream-no-annotation, response_classStreamingResponse) async def stream_story_no_annotation(): for line in message.splitlines(): yield line文档对此还有一句意味深长的总结既然 FastAPI 不介入序列化那么使用StreamingResponse就意味着你同时获得自由与责任——最终发出的每一个字节都由你自己精确地产生和编码与类型注解无关。这要求开发者对内容编码、块切分、字符集等细节心中有数。3.3 传输bytes把字符串编码后逐块 yield文档列举的首要场景之一就是发送bytes而不是str。方法与字符串完全一致只需在产出前把文本编码为字节例如 docs_src/stream_data/tutorial001_py310.py#L44-L47app.get(/story/stream-bytes, response_classStreamingResponse) async def stream_story_bytes() - AsyncIterable[bytes]: for line in message.splitlines(): yield line.encode(utf-8)AsyncIterable[bytes]这种注解对混合型数据流同样适用——你可以自由选择让yield交出字符串或字节FastAPI 不会帮你做隐式编码。示例源码文件还提供了上述两种变体在「async / 非 async」「有无注解」交叉组合下的完整版本共 8 个端点见 docs_src/stream_data/tutorial001_py310.py方便逐一对照实验。四、自定义PNGStreamingResponse用media_type子类补充Content-Type前文的示例虽然能正确流式输出数据字节但存在一个明显缺陷响应没有携带Content-Type头客户端根本不知道收到的数据是什么类型。解决办法是自定义StreamingResponse的子类通过类属性media_type指定响应头Content-Type。官方教程第二步的示例docs_src/stream_data/tutorial002_py310.py定义了一个图片流响应类class PNGStreamingResponse(StreamingResponse): media_type image/png然后在路由中把它作为response_class使用docs_src/stream_data/tutorial002_py310.py#L23-L27app.get(/image/stream, response_classPNGStreamingResponse) async def stream_image() - AsyncIterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunk这一设计有两个可见收益HTTP 层面客户端能从响应头正确识别数据类型如image/png浏览器也能直接渲染。OpenAPI 层面由于media_type已知FastAPI 生成的 OpenAPI schema 会把该接口的200响应媒体类型标记为image/png。这一点被仓库测试明确断言见下文第 8 节。4.1 用io.BytesIO在内存中模拟文件read_image()返回的是一个io.BytesIO对象——一种只存在于内存中的 file-like 对象但它提供了与真实文件一致的接口可以with打开、可以迭代逐块消费。对应定义见 docs_src/stream_data/tutorial002_py310.py#L12-L13def read_image() - BytesIO: return BytesIO(binary_image)示例中另外两个变量只是为了让演示代码能独立运行而准备的image_base64是一段 Base64 编码的小图片文本binary_image base64.b64decode(image_base64)把它解码成字节串再塞进io.BytesIO见 docs_src/stream_data/tutorial002_py310.py#L1-L13。你可以把整份文件原样复制下来直接运行无需任何外部资源。4.2 为什么用with块包裹 file-like 对象注意流式端点中with read_image() as image_file:的写法with块保证了当生成器函数结束即响应发送完毕后这个 file-like 对象会被关闭。在io.BytesIO这个具体例子里关闭与否影响不大——它是内存假文件但作者特意强调换成真实文件时确保用完后关闭文件非常重要。因此把「读取—逐块 yield」放进with上下文管理器中是处理真实文件流的推荐写法。五、文件读取与 async 的关系别让阻塞式读文件卡住 event loop大部分 file-like 对象默认并不支持 async / await它们既没有await file.read()也没有async for chunk in file这种接口。而且很多情况下读取它们属于阻塞式操作从磁盘或网络读取如果放在事件循环里直接执行会阻塞整个应用的并发处理能力。这里有个值得注意的反例上一节的io.BytesIO是例外因为数据已经全部在内存里读它不会阻塞任何东西——所以示例中即使写了async def也毫无问题。但如果面对的是真实的磁盘文件或网络流正确姿势是把path operation function声明为普通def。这样 FastAPI 会把它放进线程池threadpool的 worker 中执行yield的生成逻辑在独立线程里逐块产出从而避免阻塞主事件循环。官方示例见 docs_src/stream_data/tutorial002_py310.py#L30-L34app.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这本质上复用了 FastAPI 对同步路径操作函数的通用调度策略async def直接在事件循环运行普通def则交给线程池。当你面对「同步读取 流式输出」的组合时把整个路由声明为def是既简单又安全的选择。进阶提示如果你需要在 async 函数内部调用阻塞代码或在阻塞函数内部调用 async 代码文档推荐使用 FastAPI 的姊妹库 Asyncer 来桥接——这在混用同步 IO 与异步生成器的复杂链路中会很有用。六、用yield from跳过手写 for 循环当你正在迭代某个可迭代对象比如 file-like 对象并且打算把每个元素原样 yield 出去时Python 提供了一种简写yield from。它会替你逐个转发被迭代对象的元素从而省掉一层for循环。这不是 FastAPI 特有的语法而是纯粹的 Python 技巧但在流式响应中非常实用。官方示例 docs_src/stream_data/tutorial002_py310.py#L37-L40 演示了把整段for chunk in image_file: yield chunk压缩成一行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_fileimage_fileBytesIO本身是一个可迭代对象yield from image_file会逐个产出其中的字节块效果与逐块yield完全一致代码却更简洁。若生成器内还包含收尾逻辑例如这里的with关闭yield from也不会破坏它——文件依旧会在生成结束后被正确关闭。七、完整端点矩阵一览两个教程文件共同覆盖了「文本 / 图片文件」×「async / 非 async」×「有注解 / 无注解」×「for / yield from」的多种组合。这里汇总所有端点的路径与语义方便读者对照实验文本流示例tutorial001_py310.py路径函数形态数据形式/story/streamasync def注解AsyncIterable[str]字符串行/story/stream-no-asyncdef注解Iterable[str]字符串行/story/stream-no-annotationasync def无注解字符串行/story/stream-no-async-no-annotationdef无注解字符串行/story/stream-bytesasync def注解AsyncIterable[bytes]UTF-8 编码字节/story/stream-no-async-bytesdef注解Iterable[bytes]编码字节/story/stream-no-annotation-bytesasync def无注解编码字节/story/stream-no-async-no-annotation-bytesdef无注解编码字节图片流示例tutorial002_py310.py路径函数形态要点/image/streamasync defwithfor chunk/image/stream-no-asyncdef线程池中同步读取/image/stream-no-async-yield-fromdefyield from image_file/image/stream-no-annotationasync def无注解—/image/stream-no-async-no-annotationdef无注解—八、用仓库测试验证一切行为这两个教程文件并非孤立的代码片段仓库为它们配套了完整的自动化测试是理解行为边界的绝佳参考tests/test_tutorial/test_stream_data/test_tutorial001.py通过TestClient对/story/stream系列全部 8 个端点逐一断言每个路径都返回 HTTP200响应体文本与预期台词完全一致即多行台词被按行流式拼接后内容不丢失、不重排顺带对/openapi.json做快照断言确认各端点生成的 OpenAPI 描述符合预期没有多余的响应体 schema。tests/test_tutorial/test_stream_data/test_tutorial002.py对/image/stream系列 5 个端点断言响应状态码为200响应头content-type等于image/png——验证了自定义media_type子类确实生效响应字节内容与mod.binary_image逐字节一致——证明with、for chunk、yield from、有无注解等写法的产出完全相同同时断言 OpenAPI 中该接口的响应媒体类型被标记为image/png。这两个测试文件事实上回答了一个常见疑问「这么多等价写法结果到底一不一样」——答案是对客户端而言文本/字节内容与媒体类型完全一致不同写法主要影响的是实现风格与服务端的调度方式。九、启动与验证方式本仓库基于pyproject.toml与uv.lock管理依赖。若要在本地体验上述流式端点可参考仓库脚本 scripts/test.sh 与 scripts/lint.sh 了解测试与静态检查的执行约定并用如下常规方式手动验证以docs_src为可导入包把示例文件作为模块启动例如uvicorn docs_src.stream_data.tutorial001_py310:app --reload需在仓库根目录运行并确保 fastapi、uvicorn 等依赖已按 pyproject.toml 安装。用支持流式读取的客户端观察效果。对文本端点可用curl直观看到分块效果curl -N http://127.0.0.1:8000/story/stream-N即--no-buffer会让 curl 不做缓冲分块到达即打印便于体会「逐块下发」的过程。对图片端点验证响应头与字节完整性curl -N -D - http://127.0.0.1:8000/image/stream -o out.png观察输出中Content-Type: image/png并可用file out.png确认它确实是一张合法 PNG。若需回归验证全部端点行为直接运行仓库自带测试pytest tests/test_tutorial/test_stream_data/十、小结给流式端点写出正确代码的三个决策点综合文档与仓库源码为「原始数据流」编写 FastAPI 端点时只需要在三个决策点上做选择选方案结构化 JSON 数据 → 参考 JSON Lines 流式指南纯字符串 / 二进制 → 用本文的response_classStreamingResponse方案。定形态函数体内只有内存对象如io.BytesIO且无阻塞 IO可以用async def涉及磁盘、网络等阻塞读取用普通def交给线程池避免卡住事件循环。补语义用自定义子类的media_type让响应携带正确的Content-Type同时让 OpenAPI 文档中的媒体类型正确呈现用with块管理 file-like 对象的生命周期能用yield from的场景就少写一层循环。三个决策全部落在 tutorial001_py310.py 与 tutorial002_py310.py 两个示例文件里而正确性则由 test_tutorial001.py 与 test_tutorial002.py 钉死。按图索骥即可在真实项目中安全、高效地落地 AI 文本流、大文件与音视频流的服务端输出。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考