FastAPI 数据流式传输实战StreamingResponse 从字符串到二进制图片的完整指南【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本篇指南基于 FastAPI 官方文档中的「数据的流式传输」Stream Data章节展开讲解 FastAPI 0.134.0 新增的原始流式传输能力如何在 path operation 函数中直接用yield逐块发送字符串或二进制数据、如何为流式响应声明正确的Content-Type、以及如何安全地流式处理文件而不阻塞事件循环。读完后你将掌握用response_classStreamingResponse实现 LLM 输出直传、大文件/音视频边读边发的完整方案并理解 FastAPI 路由层对生成器端点的底层处理机制。适用场景与前置说明FastAPI 的流式能力分为两个层次先明确本篇的范围如果你要流式传输可以结构化为 JSON 的数据例如逐行输出 JSON 对象的 LLM 增量响应应使用 JSON Lines 流式传输 对应的教程本篇聚焦纯二进制数据或原始字符串的流式传输例如把 AI LLM 服务的输出原封不动地作为纯字符串发出去。注意该能力于FastAPI 0.134.0版本引入使用前请确认版本满足要求。典型的三个用例AI LLM 输出直传把 LLM 服务逐 token 产生的字符串原样转发给客户端不做任何 JSON 包装巨大二进制文件不一次性读入内存而是边读边按 chunk 发送视频与音频边处理边生成、边发送适用于媒体转码、实时合成等管道场景。用yield配合StreamingResponse在 path operation 函数上声明response_classStreamingResponse后函数体就可以是一个生成器用yield把数据按 chunk 逐次送出。官方示例位于 tutorial001_py310.pyapp.get(/story/stream, response_classStreamingResponse) async def stream_story() - AsyncIterable[str]: for line in message.splitlines(): yield line这里的关键点是FastAPI 会把每个数据 chunk 原样传给StreamingResponse不会尝试把它转换成 JSON 或做任何其他序列化。从源码可以印证这一点。StreamingResponse本身就是对 Starlette 实现的直接再导出见 fastapi/responses.pyfrom starlette.responses import StreamingResponse as StreamingResponse # noqa而在路由层的处理逻辑里当 FastAPI 检测到端点是生成器异步或同步且显式指定了response_class时走的是专门的原始流式分支见 fastapi/routing.pyelif _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: AsyncIterator[Any], ) - AsyncIterator[Any]: async for chunk in async_gen: yield chunk # To allow for cancellation to trigger await anyio.sleep(0) gen _async_stream_raw(gen) response_args _build_response_args( status_codestatus_code, solved_resultsolved_result ) response actual_response_class(contentgen, **response_args)这段代码说明两件事生成器的产出被直接包装为响应内容actual_response_class(contentgen, ...)中间不经过serialize_response的 JSON 序列化路径——这正是chunk 原样发送的底层依据对异步生成器FastAPI 额外包了一层_async_stream_raw在每个 chunk 之后await anyio.sleep(0)目的是让客户端断开连接时的取消操作能够及时生效源码注释引用了 issue #14680。非 async 的 path operation 函数不使用async的普通def函数同样可以使用yield效果完全等价app.get(/story/stream-no-async, response_classStreamingResponse) def stream_story_no_async() - Iterable[str]: for line in message.splitlines(): yield line对应源码见 tutorial001_py310.py。从上面 routing.py 的源码结构看同步生成器会直接以contentgen交给StreamingResponseStarlette 在发送时会把同步迭代器放到线程池中消费因此不会阻塞事件循环。不声明类型注解流式传输二进制数据时其实不需要声明返回值的类型注解app.get(/story/stream-no-annotation, response_classStreamingResponse) async def stream_story_no_annotation(): for line in message.splitlines(): yield line对应源码见 tutorial001_py310.py。原因很简单StreamingResponse路径下 FastAPI 不会用 Pydantic 去 JSON 化或序列化数据类型注解只是给编辑器和静态工具看的辅助信息FastAPI 本身并不消费它。由此带来一个结论在使用StreamingResponse时你拥有自由也有责任——不依赖类型注解想以什么形式发送就自己负责把数据生成并编码成相应的字节串。流式传输字节串bytes最主要的应用场景之一是流式传输bytes而不是字符串当然完全可行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。示例文件里实际提供了 8 种组合async/同步 × 有注解/无注解 × 字符串/字节官方测试 test_tutorial001.py 对这 8 个端点逐一发起请求断言status_code 200且响应文本完全一致证明这些写法在行为上完全等价。测试还验证了一个细节这类原始流式端点的 OpenAPI schema 中对应操作只声明200: Successful Response不会为yield产出的AsyncIterable[str]推断出响应体结构见 test_tutorial001.py 中的 schema 快照这也从侧面印证了 FastAPI 对这条路径不做响应模型解析。自定义PNGStreamingResponse上一节的例子流式发送了字节串但响应没有Content-Type头客户端无法识别收到的数据是什么类型。解决办法是创建一个继承StreamingResponse的自定义类按流式数据的种类设置Content-Type。例如把media_type属性设为image/png的PNGStreamingResponse见 tutorial002_py310.pyclass PNGStreamingResponse(StreamingResponse): media_type image/png然后就可以在 path operation 函数中直接把这个新类作为response_class使用app.get(/image/stream, response_classPNGStreamingResponse) async def stream_image() - AsyncIterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunk对应源码见 tutorial002_py310.py。官方测试 test_tutorial002.py 对此做了双重断言assert response.headers[content-type] image/png assert response.content mod.binary_image即响应头必须是image/png且响应体与原图字节完全一致。该测试的 OpenAPI schema 快照还显示由于自定义响应类声明了media_typeOpenAPI 中会生成content: {image/png: {schema: {type: string}}}见 test_tutorial002.py说明media_type不仅影响传输也影响接口文档。用io.BytesIO模拟文件上面的示例用io.BytesIO来模拟文件。它是只存在于内存中的文件类对象却提供与普通文件相同的接口——比如可以像文件一样迭代并读取内容image_base64 iVBORw0KGgoAAAANSUhEUgAAAB0AAAAdCAYAAABWk2cP... # 缩略展示 binary_image base64.b64decode(image_base64) def read_image() - BytesIO: return BytesIO(binary_image)对应源码见 tutorial002_py310.py。技术细节示例中的image_base64与binary_image两个变量是把一张图片 Base64 编码后解码回bytes再传给io.BytesIO得到的。这么做只是为了让示例在单文件内自包含、拷贝下来即可运行。实际项目中你会从磁盘或网络获取真实文件。使用with块的意义在于生成器函数带yield的函数结束、即响应发送完成之后该文件类对象会被确定性地关闭。对io.BytesIO这种内存对象无所谓但对真实文件而言用完后确保关闭资源是重要的工程实践。文件与非异步async多数情况下文件类对象默认与 async/await不兼容它们不提供await file.read()或async for chunk in file这类异步接口由于通常要从磁盘或网络读取读取操作是阻塞式的可能阻塞事件循环。注意上一节的io.BytesIO是例外数据已经在内存里读取不会阻塞任何东西。但大多数情况下读文件或文件类对象都会阻塞。为了避免阻塞事件循环把 path operation 函数声明为普通的def而不是async def。这样 FastAPI 会把它放到线程池 worker上执行从而避开主循环被阻塞的问题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对应源码见 tutorial002_py310.py。小贴士如果确实需要在 async 函数中调用阻塞代码或在阻塞函数中调用 async 函数可以考虑使用 Asyncer 这类桥接库它是 FastAPI 的兄弟项目。用yield from简化迭代如果你正在迭代某个文件类对象并对每个元素逐一yield可以用yield from省略显式的for循环直接把整个可迭代对象的产出转发出去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对应源码见 tutorial002_py310.py。这只是 Python 语言自身的特性并非 FastAPI 专有但确实是流式转发文件内容时很方便的小技巧。小结选型对照需求做法关键源码流式传输 JSON 可结构化数据使用 JSON Lines 流式传输教程文档原始字符串/字节流response_classStreamingResponseyieldtutorial001_py310.py需要正确Content-Type继承StreamingResponse并覆盖media_typetutorial002_py310.py读取真实文件阻塞 IO用同步def让 FastAPI 放入线程池执行tutorial002_py310.py批量转发可迭代对象yield fromtutorial002_py310.py最后两点工程提醒其一StreamingResponse路径下 FastAPI 不做任何序列化编码责任完全在开发者其二原始流式端点的 OpenAPI 不会推断出响应体 schema若需要客户端正确解析流内容如image/png务必像PNGStreamingResponse那样显式声明media_type。相关行为均可通过 tests/test_tutorial/test_stream_data/ 下的测试用例复现验证。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastAPI 数据流式传输实战：StreamingResponse 从字符串到二进制图片的完整指南
FastAPI 数据流式传输实战StreamingResponse 从字符串到二进制图片的完整指南【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本篇指南基于 FastAPI 官方文档中的「数据的流式传输」Stream Data章节展开讲解 FastAPI 0.134.0 新增的原始流式传输能力如何在 path operation 函数中直接用yield逐块发送字符串或二进制数据、如何为流式响应声明正确的Content-Type、以及如何安全地流式处理文件而不阻塞事件循环。读完后你将掌握用response_classStreamingResponse实现 LLM 输出直传、大文件/音视频边读边发的完整方案并理解 FastAPI 路由层对生成器端点的底层处理机制。适用场景与前置说明FastAPI 的流式能力分为两个层次先明确本篇的范围如果你要流式传输可以结构化为 JSON 的数据例如逐行输出 JSON 对象的 LLM 增量响应应使用 JSON Lines 流式传输 对应的教程本篇聚焦纯二进制数据或原始字符串的流式传输例如把 AI LLM 服务的输出原封不动地作为纯字符串发出去。注意该能力于FastAPI 0.134.0版本引入使用前请确认版本满足要求。典型的三个用例AI LLM 输出直传把 LLM 服务逐 token 产生的字符串原样转发给客户端不做任何 JSON 包装巨大二进制文件不一次性读入内存而是边读边按 chunk 发送视频与音频边处理边生成、边发送适用于媒体转码、实时合成等管道场景。用yield配合StreamingResponse在 path operation 函数上声明response_classStreamingResponse后函数体就可以是一个生成器用yield把数据按 chunk 逐次送出。官方示例位于 tutorial001_py310.pyapp.get(/story/stream, response_classStreamingResponse) async def stream_story() - AsyncIterable[str]: for line in message.splitlines(): yield line这里的关键点是FastAPI 会把每个数据 chunk 原样传给StreamingResponse不会尝试把它转换成 JSON 或做任何其他序列化。从源码可以印证这一点。StreamingResponse本身就是对 Starlette 实现的直接再导出见 fastapi/responses.pyfrom starlette.responses import StreamingResponse as StreamingResponse # noqa而在路由层的处理逻辑里当 FastAPI 检测到端点是生成器异步或同步且显式指定了response_class时走的是专门的原始流式分支见 fastapi/routing.pyelif _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: AsyncIterator[Any], ) - AsyncIterator[Any]: async for chunk in async_gen: yield chunk # To allow for cancellation to trigger await anyio.sleep(0) gen _async_stream_raw(gen) response_args _build_response_args( status_codestatus_code, solved_resultsolved_result ) response actual_response_class(contentgen, **response_args)这段代码说明两件事生成器的产出被直接包装为响应内容actual_response_class(contentgen, ...)中间不经过serialize_response的 JSON 序列化路径——这正是chunk 原样发送的底层依据对异步生成器FastAPI 额外包了一层_async_stream_raw在每个 chunk 之后await anyio.sleep(0)目的是让客户端断开连接时的取消操作能够及时生效源码注释引用了 issue #14680。非 async 的 path operation 函数不使用async的普通def函数同样可以使用yield效果完全等价app.get(/story/stream-no-async, response_classStreamingResponse) def stream_story_no_async() - Iterable[str]: for line in message.splitlines(): yield line对应源码见 tutorial001_py310.py。从上面 routing.py 的源码结构看同步生成器会直接以contentgen交给StreamingResponseStarlette 在发送时会把同步迭代器放到线程池中消费因此不会阻塞事件循环。不声明类型注解流式传输二进制数据时其实不需要声明返回值的类型注解app.get(/story/stream-no-annotation, response_classStreamingResponse) async def stream_story_no_annotation(): for line in message.splitlines(): yield line对应源码见 tutorial001_py310.py。原因很简单StreamingResponse路径下 FastAPI 不会用 Pydantic 去 JSON 化或序列化数据类型注解只是给编辑器和静态工具看的辅助信息FastAPI 本身并不消费它。由此带来一个结论在使用StreamingResponse时你拥有自由也有责任——不依赖类型注解想以什么形式发送就自己负责把数据生成并编码成相应的字节串。流式传输字节串bytes最主要的应用场景之一是流式传输bytes而不是字符串当然完全可行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。示例文件里实际提供了 8 种组合async/同步 × 有注解/无注解 × 字符串/字节官方测试 test_tutorial001.py 对这 8 个端点逐一发起请求断言status_code 200且响应文本完全一致证明这些写法在行为上完全等价。测试还验证了一个细节这类原始流式端点的 OpenAPI schema 中对应操作只声明200: Successful Response不会为yield产出的AsyncIterable[str]推断出响应体结构见 test_tutorial001.py 中的 schema 快照这也从侧面印证了 FastAPI 对这条路径不做响应模型解析。自定义PNGStreamingResponse上一节的例子流式发送了字节串但响应没有Content-Type头客户端无法识别收到的数据是什么类型。解决办法是创建一个继承StreamingResponse的自定义类按流式数据的种类设置Content-Type。例如把media_type属性设为image/png的PNGStreamingResponse见 tutorial002_py310.pyclass PNGStreamingResponse(StreamingResponse): media_type image/png然后就可以在 path operation 函数中直接把这个新类作为response_class使用app.get(/image/stream, response_classPNGStreamingResponse) async def stream_image() - AsyncIterable[bytes]: with read_image() as image_file: for chunk in image_file: yield chunk对应源码见 tutorial002_py310.py。官方测试 test_tutorial002.py 对此做了双重断言assert response.headers[content-type] image/png assert response.content mod.binary_image即响应头必须是image/png且响应体与原图字节完全一致。该测试的 OpenAPI schema 快照还显示由于自定义响应类声明了media_typeOpenAPI 中会生成content: {image/png: {schema: {type: string}}}见 test_tutorial002.py说明media_type不仅影响传输也影响接口文档。用io.BytesIO模拟文件上面的示例用io.BytesIO来模拟文件。它是只存在于内存中的文件类对象却提供与普通文件相同的接口——比如可以像文件一样迭代并读取内容image_base64 iVBORw0KGgoAAAANSUhEUgAAAB0AAAAdCAYAAABWk2cP... # 缩略展示 binary_image base64.b64decode(image_base64) def read_image() - BytesIO: return BytesIO(binary_image)对应源码见 tutorial002_py310.py。技术细节示例中的image_base64与binary_image两个变量是把一张图片 Base64 编码后解码回bytes再传给io.BytesIO得到的。这么做只是为了让示例在单文件内自包含、拷贝下来即可运行。实际项目中你会从磁盘或网络获取真实文件。使用with块的意义在于生成器函数带yield的函数结束、即响应发送完成之后该文件类对象会被确定性地关闭。对io.BytesIO这种内存对象无所谓但对真实文件而言用完后确保关闭资源是重要的工程实践。文件与非异步async多数情况下文件类对象默认与 async/await不兼容它们不提供await file.read()或async for chunk in file这类异步接口由于通常要从磁盘或网络读取读取操作是阻塞式的可能阻塞事件循环。注意上一节的io.BytesIO是例外数据已经在内存里读取不会阻塞任何东西。但大多数情况下读文件或文件类对象都会阻塞。为了避免阻塞事件循环把 path operation 函数声明为普通的def而不是async def。这样 FastAPI 会把它放到线程池 worker上执行从而避开主循环被阻塞的问题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对应源码见 tutorial002_py310.py。小贴士如果确实需要在 async 函数中调用阻塞代码或在阻塞函数中调用 async 函数可以考虑使用 Asyncer 这类桥接库它是 FastAPI 的兄弟项目。用yield from简化迭代如果你正在迭代某个文件类对象并对每个元素逐一yield可以用yield from省略显式的for循环直接把整个可迭代对象的产出转发出去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对应源码见 tutorial002_py310.py。这只是 Python 语言自身的特性并非 FastAPI 专有但确实是流式转发文件内容时很方便的小技巧。小结选型对照需求做法关键源码流式传输 JSON 可结构化数据使用 JSON Lines 流式传输教程文档原始字符串/字节流response_classStreamingResponseyieldtutorial001_py310.py需要正确Content-Type继承StreamingResponse并覆盖media_typetutorial002_py310.py读取真实文件阻塞 IO用同步def让 FastAPI 放入线程池执行tutorial002_py310.py批量转发可迭代对象yield fromtutorial002_py310.py最后两点工程提醒其一StreamingResponse路径下 FastAPI 不做任何序列化编码责任完全在开发者其二原始流式端点的 OpenAPI 不会推断出响应体 schema若需要客户端正确解析流内容如image/png务必像PNGStreamingResponse那样显式声明media_type。相关行为均可通过 tests/test_tutorial/test_stream_data/ 下的测试用例复现验证。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考