Litestar 流式响应完全指南Stream 与 ASGIStreamingResponse 的实战与原理【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar流式响应Streaming Response允许服务端不等待全部数据生成完毕而是将数据分块、持续地推送给客户端是构建实时时钟、长轮询、大文件下载、日志流水与 AI 推理输出等场景的核心能力。本文以 Litestar 的Stream与ASGIStreamingResponse为线索结合 streaming.py 源码与测试用例讲解从路由返回流式响应到 ASGI 底层分块发送的完整链路帮助你掌握同步/异步生成器、参数配置、断连取消与 HEAD 请求等关键细节。一、何时需要流式响应与一次性渲染整个响应体的普通Response不同流式响应把「数据生产」与「数据发送」解耦。典型适用场景包括实时数据推送如不断刷新的时间戳、股票行情、日志流大文件传输文件超过 1 MB 时见 responses.rst 中关于文件响应的说明分块下发避免整块载入内存长耗时计算边计算边返回中间结果缩短首字节时间TTFB第三方流透传将上游 API 的流式响应原样转发给客户端。Litestar 为流式响应提供了两个层面的类类层级职责Stream高层 API在路由处理器中直接返回接受任意同步/异步可迭代对象负责转换为 ASGI 响应ASGIStreamingResponse底层 ASGI真正执行http.response.body事件的分块发送、断连监听两者都定义在 litestar/response/streaming.py通过__all__对外暴露其余响应类File、Redirect、SSE、Template等的索引可参见 docs/reference/response/index.rst。二、快速上手在路由中返回 Stream官方示例 streaming_responses.py 给出了最简用法——一个每隔 10 毫秒产出一次当前时间的无限流from asyncio import sleep from collections.abc import AsyncGenerator from datetime import datetime from litestar import Litestar, get from litestar.response import Stream from litestar.serialization import encode_json async def my_generator() - AsyncGenerator[bytes, None]: while True: await sleep(0.01) yield encode_json({current_time: datetime.now()}) get(path/time, sync_to_threadFalse) def stream_time() - Stream: return Stream(my_generator()) app Litestar(route_handlers[stream_time])关键点Stream的第一个位置参数必须是可迭代对象此处为my_generator()的实例生成器产出的是已序列化好的bytes这里用encode_json手动编码也可以直接 yieldstrsync_to_threadFalse表示处理器本身是纯同步函数、不需要丢到线程池执行若处理器内部有阻塞 IO 则应保留默认的线程池调度。启动应用后访问/time浏览器或curl会持续收到{current_time: ...}形式的 JSON 块直到连接关闭。三、Stream 的参数全解Stream.__init__的完整签名源码见 streaming.pyStream( content: StreamType[str | bytes] | Callable[[], StreamType[str | bytes]], *, backgroundNone, cookiesNone, encodingutf-8, headersNone, media_typeNone, status_codeNone, )各参数语义如下参数类型默认值说明content可迭代对象或返回可迭代对象的可调用对象必填流的数据源见下文「可接受的迭代器形态」backgroundBackgroundTask/BackgroundTasksNone响应发送完成后执行的后台任务cookiesCookie列表None写入响应Set-Cookie头的 Cookie 集合encodingstrutf-8响应头编码同时用于把str分块编码为bytesheaders字符串键字典None自定义响应头键名大小写不敏感media_typeMediaType/OpenAPIMediaType/strNone最终回退到 JSON写入Content-Type头的值status_codeintNone默认 200HTTP 状态码其中content的形态非常灵活。官方文档 responses.rst 特别说明可以是返回同步/异步生成器的可调用对象、生成器本身、同步/异步迭代器类或同步/异步迭代器类的实例。类型别名StreamType定义在 helper_types.pyStreamType: TypeAlias Union[Iterable[T], Iterator[T], AsyncIterable[T], AsyncIterator[T]]Stream.to_asgi_responsestreaming.py在转换时对content做了归一化处理iterator self.iterator if not isinstance(iterator, (Iterable, Iterator, AsyncIterable, AsyncIterator)) and callable(iterator): iterator iterator()即传可调用对象时会在请求发生时调用它取得真正的迭代器。这保证了每次请求都获得全新的迭代器避免复用已耗尽exhausted的生成器。四、ASGIStreamingResponse底层如何把流发出去Stream本身并不发送数据它通过to_asgi_response将自身转换为ASGIStreamingResponsestreaming.py。转换过程的关键逻辑iterator self.iterator if not isinstance(iterator, (Iterable, Iterator, AsyncIterable, AsyncIterator)) and callable(iterator): iterator iterator() return ASGIStreamingResponse( backgroundself.background or background, content_length0, cookiesself._merge_cookies(cookies), encodingself.encoding, headersheaders, is_head_responseis_head_response, iteratoriterator, media_typemedia_type, status_codeself.status_code or status_code, )这里有两个容易忽略的细节content_length0流式响应在发送前不知道总长度因此不会设置content-length头。对应测试 test_streaming_response_unknown_size.py 断言了「不包含 content-length 头」这一行为如果你确实知道总长可以在构造ASGIStreamingResponse时通过headers{content-length: 10}显式指定测试 test_streaming_response_known_size.py 覆盖了该场景。同步迭代器的异步化ASGIStreamingResponse.__init__会用AsyncIteratorWrapper包装同步可迭代对象streaming.py。该包装器定义在 sync.py通过sync_to_thread将每次next()调用放到线程池执行从而不阻塞事件循环——这是同步生成器能够以异步方式流式发送的底层支撑。ASGIStreamingResponse自身的构造函数还接收is_head_responseHEAD 请求时不发送 body以及继承自ASGIResponse的通用参数media_type、status_code、cookies、encoding、headers、background。基类ASGIResponse位于 base.py其__call__流程为start_response发送http.response.start→send_body发送 body→after_response执行后台任务。4.1 分块发送_stream 方法_streamstreaming.py逐块从迭代器取值并编码发送async for chunk in self.iterator: stream_event: HTTPResponseBodyEvent { type: http.response.body, body: chunk if isinstance(chunk, bytes) else chunk.encode(self.encoding), more_body: True, } await send(stream_event) terminus_event: HTTPResponseBodyEvent {type: http.response.body, body: b, more_body: False} await send(terminus_event)要点每个分块都以more_bodyTrue发送告知 ASGI 服务器「后面还有数据」str分块按encoding默认utf-8编码bytes分块原样透传迭代耗尽后发送一个空 body、more_bodyFalse的终结事件标志流结束。4.2 断连取消send_body 与 _listen_for_disconnect无限流最怕客户端断开后服务器还在空转。send_bodystreaming.py通过 anyio 任务组同时做两件事async with create_task_group() as task_group: task_group.start_soon(partial(self._stream, send)) await self._listen_for_disconnect(cancel_scopetask_group.cancel_scope, receivereceive)_listen_for_disconnectstreaming.py持续监听 ASGI 的receive通道一旦收到http.disconnect消息就调用cancel_scope.cancel()终止整个任务组从而立即停止迭代器的消费。注意源码注释特别提醒anyio 3 中cancel_scope.cancel()不是协程不能await。对应测试 test_streaming_response.py 分别用异步迭代器与同步迭代器cycle([1, 2, 3])验证即使迭代器理论上无限产出收到断连后流也会自行停止不会挂起。4.3 自定义迭代器与后台任务测试用例还演示了三种官方支持的可迭代形态自定义异步迭代器类实现__aiter__/__anext__以StopAsyncIteration结束test_streaming_response.py自定义异步可迭代类仅实现__aiter__test_streaming_response.py同步生成器直接传给ASGIStreamingResponse由内部包装为异步test_streaming_response.py。此外 test_streaming_response.py 展示了background参数流发送完成后BackgroundTask会接着执行示例中把「6, 7, 8, 9」填充到外部变量实现「先流式返回、再异步收尾」的清理逻辑。五、与 File、SSE 等响应类型的边界Litestar 的响应体系在 docs/reference/response/index.rst 中统一索引各类型定位不同Stream通用字节流/字符串流是本文主题File文件响应超过默认 1 MB 的chunk_size时同样以分块方式流式发送见 responses.rst并支持 fsspec 文件系统协议ServerSentEventSSE面向浏览器的服务器推送事件协议基于文本流详见 sse.rst 与示例 sse_responses.pyRedirect、Template分别处理跳转与模板渲染均不涉及流式传输。选择建议若只是透传/推送任意分块数据用Stream若是文件下载用File自动处理分块与文件系统若是浏览器实时事件推送用SSE。六、实战注意事项HEAD 请求流式响应配合 HEAD 请求时is_head_responseTrue会跳过 body 发送仅返回头部信息base.py。编码一致性encoding参数同时影响响应头与str分块的编码多语言内容务必显式指定避免乱码。媒体类型回退media_type在Stream、to_asgi_response均未指定时最终回退为MediaType.JSON见 streaming.py并可能影响Content-Type的拼接格式。生产环境代理缓冲无限流经过 Nginx 等反向代理时可能需要关闭缓冲如proxy_buffering off才能即时透传这属于部署层配置。资源释放若迭代器持有文件句柄或数据库游标推荐把资源释放逻辑放进finally块或利用background参数做流结束后的清理——断连取消见 4.2 节同样会触发迭代器的异常退出路径。七、延伸阅读响应基类与通用渲染逻辑base.py流式响应核心实现streaming.py官方使用指南Streaming Responses 小节responses.rst可运行示例streaming_responses.py单元测试断连取消、自定义迭代器、后台任务test_streaming_response.pyStreamType类型别名定义helper_types.py【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
