Starlette 响应体系全解：从 Response 基类到流式与文件响应的源码级实践

Starlette 响应体系全解：从 Response 基类到流式与文件响应的源码级实践

【免费下载链接】starletteThe little ASGI framework that shines. 🌟项目地址: https://gitcode.com/gh_mirrors/st/starlette

Starlette 提供了一套完整的响应类家族，统一通过向 ASGI 的send通道发送http.response.start与http.response.body消息来返回 HTTP 响应。本文以 docs/responses.md 为主线，结合 starlette/responses.py 源码与 tests/test_responses.py 测试用例，逐一剖析Response、HTMLResponse、PlainTextResponse、JSONResponse、RedirectResponse、StreamingResponse、FileResponse的设计与实战要点，读完即可掌握：如何构造带自动响应头的响应、如何操作 Cookie、如何定制 JSON 序列化、如何流式传输内容、如何利用文件响应的 Range 断点续传能力，以及如何挂载后台任务。

响应类总览：一切皆可被调用的 ASGI 应用

在 Starlette 中，所有响应对象都实现了__call__(scope, receive, send)方法，因此响应实例本身就是一个合法的 ASGI 应用。你可以在async def app(scope, receive, send)中直接await response(scope, receive, send)把它发送出去，也可以把响应对象直接交给路由、TestClient 或任何 ASGI 服务器执行。

Response基类位于 starlette/responses.py，其发送流程在__call__中一目了然（源码 L163-L170）：

async def __call__(self, scope, receive, send): if scope["type"] == "websocket": send = self._wrap_websocket_denial_send(send) await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers}) await send({"type": "http.response.body", "body": self.body}) if self.background is not None: await self.background()

这段代码还透露了两个重要细节：

• WebSocket 场景的拒绝响应：当 scope 类型为websocket时，响应消息会被_wrap_websocket_denial_send改写为websocket.http.response.start/websocket.http.response.body，从而向客户端发送 HTTP 拒绝响应（如 403/405），这是 WebSocket 握手失败时的标准反馈机制；
• 后台任务：响应体发送完毕后，若构造时传入了background（来自 starlette/background.py 的BackgroundTask），会继续执行该任务——这是"响应后清理/记账/通知"等场景的官方挂载点。

Response 基类：签名与自动响应头

Response的构造签名与文档一致：

Response(content, status_code=200, headers=None, media_type=None)

• content：字符串或字节串（源码render同时支持bytes | memoryview，L48-L53）；传入None时渲染为空字节串；
• status_code：整数 HTTP 状态码，默认200；
• headers：字符串字典；
• media_type：媒体类型字符串，例如"text/html"。

此外源码中还有一个文档之外的隐藏参数background: BackgroundTask | None = None（L39），用于挂载后台任务，前文已述。

Content-Length 与 Content-Type 的自动填充

init_headers（L55-L81）负责生成响应头，规则如下：

• Content-Length 自动计算：基于渲染后的self.body长度。但有三个例外不会填充Content-Length：状态码小于 200（如 1xx）、状态码为204（No Content）、状态码为304（Not Modified）；
• Content-Type 自动推导：基于media_type；若媒体类型以text/开头且尚未显式包含charset=，则自动追加; charset=utf-8（Response.charset类属性默认"utf-8"）；
• 去重保护：如果你在headers中已显式提供content-length或content-type，Starlette 不会重复生成，以你传入的值为准；
• 大小写规范化：传入的 headers 键会被转成小写并以latin-1编码保存为raw_headers（ASGI 要求的list[tuple[bytes, bytes]]格式）。

以下是最小完整示例（文档原样保留，可直接作为独立 ASGI 应用运行）：

from starlette.responses import Response async def app(scope, receive, send): assert scope['type'] == 'http' response = Response('Hello, world!', media_type='text/plain') await response(scope, receive, send)

对应测试见 test_text_response 与 test_bytes_response：字符串内容返回response.text，字节内容（如media_type="image/png"）返回response.content。

动态修改响应头

响应对象还提供headers属性（MutableHeaders），发送前可随时增删改：

response = Response("hello, world", media_type="text/plain", headers={"x-header-1": "123"}) response.headers["x-header-2"] = "789"

测试 test_response_headers 验证了这种先构造后修改的用法，两个自定义头都会出现在最终响应中。

Cookie 操作：set_cookie 与 delete_cookie

Starlette 在响应对象上提供了set_cookie方法（L89-L132），底层基于标准库http.cookies.SimpleCookie生成符合规范的Set-Cookie头，并直接追加到raw_headers。

完整签名如下：

Response.set_cookie(key, value, max_age=None, expires=None, path="/", domain=None, secure=False, httponly=False, samesite="lax", partitioned=False)

关于samesite，源码 L120-L124 中有硬性断言，非法值会在发送前直接AssertionError；关于partitioned，源码 L126-L129 在sys.version_info < (3, 14)时抛出ValueError("Partitioned cookies are only supported in Python 3.14 and above.")，测试 test_set_cookie_raises_for_invalid_python_version 也针对 <3.14 环境做了跳过与校验。

delete_cookie：快速作废 Cookie

delete_cookie是对set_cookie的便捷封装（L134-L152），它将max_age=0、expires=0重新调用set_cookie，从而让浏览器立即过期并删除该 Cookie：

Response.delete_cookie(key, path='/', domain=None)

签名中同样可以透传secure、httponly、samesite参数（源码实现默认沿用lax）。对应测试见 test_delete_cookie。

HTMLResponse 与 PlainTextResponse：最常用的两个快捷类

这两个类都是Response的极简子类，唯一的区别是预设了media_type类属性（L173-L178）：

• HTMLResponse：media_type = "text/html"，接收文本或字节，返回 HTML 页面；
• PlainTextResponse：media_type = "text/plain"，返回纯文本。

from starlette.responses import HTMLResponse async def app(scope, receive, send): assert scope['type'] == 'http' response = HTMLResponse('<html><body><h1>Hello, world!</h1></body></html>') await response(scope, receive, send)

from starlette.responses import PlainTextResponse async def app(scope, receive, send): assert scope['type'] == 'http' response = PlainTextResponse('Hello, world!') await response(scope, receive, send)

由于media_type以text/开头，基类的init_headers会自动补上charset=utf-8。这也解释了为什么PlainTextResponse常被框架内部用于返回错误文本——例如 starlette/routing.py 的 404 响应、L279 的 405 Method Not Allowed 响应，以及 FileResponse 内部的 400/416 错误响应。

JSONResponse：默认序列化与自定义 render

JSONResponse接收任意可被 JSON 序列化的数据，返回application/json编码的响应（L181-L201）。

from starlette.responses import JSONResponse async def app(scope, receive, send): assert scope['type'] == 'http' response = JSONResponse({'hello': 'world'}) await response(scope, receive, send)

其render方法揭示了默认序列化策略（L194-L201）：

def render(self, content: Any) -> bytes: return json.dumps( content, ensure_ascii=False, allow_nan=False, indent=None, separators=(",", ":"), ).encode("utf-8")

要点：

• ensure_ascii=False：非 ASCII 字符（如中文）不会被转义为\uXXXX，直接以 UTF-8 输出，体积更小、可读性更好；
• allow_nan=False：序列化时遇到NaN/Infinity会直接抛出ValueError，防止生成非法 JSON；
• separators=(",", ":")：紧凑输出，去掉多余空格；
• 测试 test_json_none_response 验证了JSONResponse(None)会正确返回字面量b"null"，即"空响应"也是合法 JSON。

自定义 JSON 序列化：子类化并覆写 render

当默认的json.dumps不够用时——例如要序列化datetime、UUID等非标准对象，或想换用更快的序列化库——最佳实践是子类化JSONResponse并覆写render方法。文档给出的 orjson 示例（此处 orjson 为第三方包，需自行安装）：

from typing import Any import orjson from starlette.responses import JSONResponse class OrjsonResponse(JSONResponse): def render(self, content: Any) -> bytes: return orjson.dumps(content)

render的职责是把任意content变成bytes，只要返回类型正确，基类的Content-Length计算、响应头生成等逻辑都会照常工作。不过需要说明的是：在绝大多数场景下应优先使用默认JSONResponse，只有在微优化某个特定接口、或确实需要序列化非标准对象类型时，才值得引入自定义 render。

RedirectResponse：默认 307 与 URL 编码

RedirectResponse返回 HTTP 重定向，默认状态码为307 Temporary Redirect（L204-L213）。307 与 302 的关键区别在于：307 会保留原始请求的 HTTP 方法与请求体，适合 POST 等非 GET 方法的重定向场景。

from starlette.responses import PlainTextResponse, RedirectResponse async def app(scope, receive, send): assert scope['type'] == 'http' if scope['path'] != '/': response = RedirectResponse(url='/') else: response = PlainTextResponse('Hello, world!') await response(scope, receive, send)

实现要点：

• 构造时以空内容初始化（content=b""），因此响应体的Content-Length为0，测试 test_redirect_response_content_length_header 专门验证了这一点；
• Location头由quote(str(url), safe=":/%#?=@[]!$&'()*+,;")生成（L213），保留 URL 中的合法保留字符。测试 test_quoting_redirect_response 验证了/I ♥ Starlette/会被正确编码为/I%20%E2%99%A5%20Starlette/；
• url参数接受字符串或URL对象（来自 starlette/datastructures.py）。注意：RedirectResponse没有独立的media_type参数，若需要自定义媒体类型需通过headers传入。

另外，RedirectResponse在框架内部也有应用：starlette/routing.py 会在末尾斜杠重定向（redirect_slashes=True）时构造RedirectResponse(url=str(redirect_url))，把客户端导流到带斜杠的规范化路径。

StreamingResponse：流式传输异步或同步迭代器

StreamingResponse接受异步生成器/异步迭代器或普通生成器/迭代器，将响应体分块流式发送（L222-L283），适用于 SSE、大文件导出、长轮询、逐块生成的 HTML 等"边生成边发送"的场景。

文档示例（每个数字之间延迟 0.5 秒逐步推送）：

from starlette.responses import StreamingResponse import asyncio async def slow_numbers(minimum, maximum): yield '<html><body><ul>' for number in range(minimum, maximum + 1): yield '<li>%d</li>' % number await asyncio.sleep(0.5) yield '</ul></body></html>' async def app(scope, receive, send): assert scope['type'] == 'http' generator = slow_numbers(1, 10) response = StreamingResponse(generator, media_type='text/html') await response(scope, receive, send)

同步迭代器与 file-like 对象

源码构造逻辑（L233-L236）会判断传入内容是否为AsyncIterable：

• 是 → 直接作为body_iterator；
• 否（同步迭代器）→ 用iterate_in_threadpool包装（来自 starlette/concurrency.py），把同步迭代放到线程池中执行，避免阻塞事件循环。

因此，file-like 对象（例如open()返回的文件对象）本质上是普通迭代器，可以直接传给StreamingResponse进行流式读取。测试 test_sync_streaming_response 验证了同步生成器同样可用；test_streaming_response_custom_iterator 与 test_streaming_response_custom_iterable 则验证了任意实现了__aiter__/__anext__的自定义异步迭代器/可迭代对象均可作为内容源。

流式发送协议细节

stream_response方法（L248-L255）是流式发送的核心：

• 先发送http.response.start；
• 逐块迭代：非bytes的块会自动用charset（默认 utf-8）编码，并以more_body=True标记"还有后续"；
• 最后发送空的http.response.body且more_body=False收尾。

由于响应体不是一次性渲染的，基类的自动Content-Length不会生效（没有完整的 body 可计算长度），流式响应默认使用 chunked 传输。

客户端断连与 ASGI 版本适配

源码针对 ASGI 协议版本做了分叉处理（L265-L283）：

• 当asgi.spec_version >= (2, 4)：流式发送中若抛出OSError（典型如对端已断开导致的写失败），会转换为ClientDisconnect异常（定义于 starlette/requests.py），供上层精确捕获处理；
• 更早的协议版本：通过create_collapsing_task_group（来自 starlette/_utils.py）同时监听http.disconnect消息并推进响应流，一旦任一方完成即取消另一任务组。

这保证了大响应流式推送时，客户端中途断开不会让服务器悬挂。StreamingResponse同样支持background后台任务，测试 test_streaming_response 验证了响应体（"1, 2, 3, 4, 5"）与后台任务（"6, 7, 8, 9"）各自独立完成。

FileResponse：异步文件流式响应与 HTTP Range 断点续传

FileResponse异步流式传输文件，其构造参数与其他响应类型不同（L296-L329）：

from starlette.responses import FileResponse async def app(scope, receive, send): assert scope['type'] == 'http' response = FileResponse('statics/favicon.ico') await response(scope, receive, send)

自动生成的响应头

文件响应会自动包含四类头（set_stat_headers，L331-L339）：

• Content-Length：来自stat_result.st_size；
• Last-Modified：来自文件 mtime，格式化为 HTTP-date；
• ETag：由md5(str(st_mtime) + "-" + str(st_size))生成，即"修改时间 + 文件大小"的哈希，文件未变则 ETag 不变；
• Accept-Ranges: bytes：在__init__中通过setdefault添加（L319），声明支持字节范围请求。

Content-Type则由传入的media_type或按文件名推断；若filename非空，还会生成Content-Disposition头（L320-L326）：ASCII 安全时输出attachment; filename="example.png"，含非 ASCII 字符（如中文文件名你好.txt）时输出 RFC 5987 风格的attachment; filename*=utf-8''...，对应测试见 test_file_response 与 test_file_response_with_chinese_filename。

运行时文件校验

如果构造时未提供stat_result，__call__会在线程池中执行os.stat并填充上述头（L348-L359），同时做两项校验：

• 文件不存在 →RuntimeError(f"File at path {self.path} does not exist.")；
• 路径不是普通文件（如目录）→RuntimeError(f"File at path {self.path} is not a file.")。

测试 test_file_response_with_missing_file_raises_error 与 test_file_response_with_directory_raises_error 分别覆盖了这两种异常。

HTTP Range 请求：206 / 416 与多段响应

FileResponse完整支持 HTTP 范围请求，这是文档特别强调的能力：

• 请求带Range头且文件存在时，返回206 Partial Content，只传输请求的字节区间；
• 范围不合法（如起始位置超出文件大小）时，返回416 Range Not Satisfiable，并附带Content-Range: bytes */{file_size}头（L372-L374）；
• 同时支持If-Range条件判断（L363-L365）：只有当If-Range值等于当前的Last-Modified或ETag时才处理 Range，否则返回完整文件（_should_use_range，L456-L457）。

Range 处理逻辑（__call__的 L361-L382）分支如下：

1. 无Range头（或有If-Range但条件不满足）→ 完整文件响应（_handle_simple）；
2. Range头解析失败 → 400（MalformedRangeHeader）；
3. 单个有效范围 →206单段响应（_handle_single_range），重写Content-Range: bytes {start}-{end-1}/{file_size}与新的Content-Length（L401-L418）；
4. 多个范围 →206且Content-Type变为multipart/byteranges; boundary=...的多段响应（_handle_multiple_ranges，L420-L454），boundary 由token_hex(13)生成，每段携带独立的Content-Type与Content-Range。

范围解析规则（_parse_range_header/_parse_ranges，L459-L528）还有几个工程细节：

• 只支持bytes单位，其他单位返回 400；
• 请求段数超过max_ranges = 100时直接忽略 Range、返回完整文件；
• 支持-500（末尾 500 字节）这类无起始位置的后缀范围语法；
• 多个重叠范围会被合并，避免重复传输；
• 空段（如-）与非数字段会被忽略。

此外FileResponse对两类特殊请求做了优化：

• HEAD 请求：只发送响应头、不发送正文（send_header_only，L343 与 L389-L390），测试 test_file_response_on_head_method 验证了响应头完整而 body 为空；
• pathsend 扩展：若 scope 的extensions中声明了http.response.pathsend（ASGI 服务器支持零拷贝发送），则发送http.response.pathsend消息直接交付文件路径（L344 与 L391-L392），避免应用层逐块拷贝。

文件读取以chunk_size = 64 * 1024（64KB）分块进行，且全程通过anyio.open_file异步完成，不阻塞事件循环。

第三方响应：EventSourceResponse（Server-Sent Events）

除内置响应类外，Starlette 生态中还有第三方实现的响应类，典型代表是sse-starlette 提供的EventSourceResponse（第三方包，使用时需单独安装sse-starlette）。它实现了 Server-Sent Events（SSE）协议，让服务器可以向客户端持续推送事件流——与 WebSocket 相比，SSE 基于单向 HTTP 长连接，无需额外协议握手、天然支持重连与事件 ID，适合实时通知、日志推送、AI 流式对话等"服务器单向推送"场景。

从实现角度可以理解为：EventSourceResponse本质上是一个基于StreamingResponse之上的封装，它负责把事件数据按 SSE 规范（data:/event:/id:/retry:行格式）编码成流式分块发送，同时设置正确的Content-Type: text/event-stream与禁用缓冲的相关头。若你的应用已经用到了StreamingResponse的分块机制，再理解 SSE 流式推送就会非常顺理成章。

综合实践：一个整合全部响应能力的路由示例

把文档与源码中的要点组合起来，一个同时展示 JSON、Cookie、重定向、流式与文件响应的应用可以这样组织：

from starlette.responses import ( FileResponse, JSONResponse, RedirectResponse, Response, StreamingResponse, ) async def app(scope, receive, send): assert scope['type'] == 'http' path = scope['path'] if path == '/json': response = JSONResponse({"message": "你好，Starlette"}) # UTF-8 直出，紧凑序列化 response.set_cookie("session", "abc123", max_age=3600, httponly=True, samesite="lax") elif path == '/old': response = RedirectResponse(url='/json') # 默认 307 elif path == '/stream': async def gen(): for i in range(5): yield f"chunk-{i}\n" response = StreamingResponse(gen(), media_type="text/plain") elif path == '/download': response = FileResponse('statics/example.txt', filename='example.txt') # 自动 ETag / Last-Modified / Range else: response = Response('Not Found', status_code=404, media_type='text/plain') await response(scope, receive, send)

可以借助 tests/test_responses.py 中对应的测试（test_redirect_response、test_set_cookie、test_file_response、test_streaming_response等）来验证各响应类的实际输出头与状态码，从而把文档描述落到可观测的代码行为上。

总结

Starlette 的响应体系围绕"响应即 ASGI 应用"这一核心设计展开：Response基类负责Content-Length/Content-Type的自动填充与 Cookie 操作，HTMLResponse/PlainTextResponse/JSONResponse提供最常用的快捷封装，RedirectResponse以 307 完成安全重定向，StreamingResponse打通异步与同步迭代器的分块传输，而FileResponse则把静态文件服务、ETag 缓存校验、Range 断点续传、多段 multipart 响应与零拷贝发送集于一身。无论你在框架层面使用哪种响应，都可以追溯到 starlette/responses.py 中的统一send协议实现，并用 tests/test_responses.py 中的测试用例作为行为基准。

【免费下载链接】starletteThe little ASGI framework that shines. 🌟项目地址: https://gitcode.com/gh_mirrors/st/starlette