企业官网建设流程全解析

Litestar 连接层（connection）API 全解析：ASGIConnection、Request 与 WebSocket 的底层原理与实战

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar

Litestar 的litestar.connection模块是所有 HTTP 请求与 WebSocket 连接处理的核心基座，它把 ASGI Scope/Receive/Send 协议封装成开发者友好的ASGIConnection、Request和WebSocket三个类。本文以 docs/reference/connection.rst 为骨架，结合 litestar/connection/ 包源码与 tests/unit/test_connection/ 测试，系统讲解这三个连接类的每个属性与方法，帮助你理解 Litestar 的路由处理、依赖注入、鉴权、Session 与 WebSocket 通信是如何在连接层之上构建的。

模块概览：三个类各司其职

litestar.connection包的对外导出非常精简，全部公开符号只有三个类（见 litestar/connection/init.py）：

from litestar.connection.base import ASGIConnection from litestar.connection.request import Request from litestar.connection.websocket import WebSocket __all__ = ("ASGIConnection", "Request", "WebSocket")

三者共享同一套scope、receive、send三件套构造参数，默认值分别为empty_receive与empty_send——这两个占位函数在被调用时会直接抛出RuntimeError，用于阻止在不需要收发的场景（例如仅读取 Scope 数据）下误用网络通道（litestar/connection/base.py）。

值得注意的继承细节：Request的泛型参数把HandlerT绑定为HTTPRouteHandler，WebSocket则绑定为WebsocketRouteHandler，因此connection.route_handler属性的静态类型会自动收窄为对应的处理器类型，方便直接访问路由处理器上的配置（如route_handler.type_decoders、route_handler.request_max_body_size）。

ASGIConnection：所有连接的公共基座

ASGIConnection使用__slots__声明了仅有的几个实例字段，内存占用可控；其余能力全部通过属性（property）惰性计算并借助ScopeState进行连接级缓存（litestar/connection/base.py）。

三个原始协议对象：scope / receive / send

• scope：ASGI 连接作用域字典，其中litestar_app、route_handler、path_params、state等键由 Litestar 框架在路由分发阶段写入；
• receive：异步接收函数，用于读取 ASGI 事件（如http.request、websocket.receive）；
• send：异步发送函数，用于写出 ASGI 事件（如http.response.start、websocket.send）。

常用只读属性一览

auth / user / session 的防护性设计：三个属性在 scope 中找不到对应键时都会抛出ImproperlyConfiguredException，并给出明确的提示文案——例如auth会提示"install an AuthMiddleware to set it"，session会提示"install a SessionMiddleware to set it"（litestar/connection/base.py）。这意味着这些属性不是"默认返回 None"，而是强制要求通过对应的中间件显式注入，从源头避免静默的空值传递。

base_url 的构造技巧：与url直接解析原始 scope 不同，base_url会先构造一个副本 scope——把path置为"/"、清空query_string、并用app_root_path或root_path覆盖root_path——再调用URL.from_scope，从而得到 "scheme://host[:port]/root_path" 这样的纯基础地址，方便后续拼绝对 URL（litestar/connection/base.py）。

惰性缓存机制：URL、base_url、query_params、cookies 等解析结果都会写入连接对应的ScopeState（url、base_url、parsed_query、cookies等槽位），首次访问后后续访问直接命中缓存。这一点在 tests/unit/test_connection/test_base.py 中有直接验证：访问connection.url前后，ScopeState.url从Empty变为非空。

会话管理：set_session / clear_session / get_session_id

• set_session(value)：把字典（或 pydantic 等数据容器、或Empty）写入scope["session"]；若启用了SessionMiddleware，后续响应阶段会据此生成 Cookie 头；
• clear_session()：把scope["session"]置为Empty并清空ScopeState.session_id，使中间件感知到"会话已清空"；
• get_session_id()：返回ScopeState.session_id，未设置时返回None。

反向 URL 生成：url_for

url_for(name, **path_parameters)是连接层最重要的工具方法之一：它通过app.route_reverse(name, **path_parameters)将路由处理器或路由名解析为路径，再用make_absolute_url(path, self.base_url)拼成绝对 URL（litestar/connection/base.py）。当路由名不存在、路径参数缺失或类型错误时抛出NoRouteMatchFoundException。注意make_absolute_url会先对base_url的 path 做rstrip("/")，避免出现双斜杠。

Request：HTTP 请求的完整封装

Request在ASGIConnection基础上增加了 HTTP 协议相关的全部能力（litestar/connection/request.py），并新增两个实例标志位：is_connected（正文流是否仍可读取）与supports_push_promise（ASGI 服务器是否支持http.response.push扩展，检测逻辑见 enums.py 中的ASGIExtension.SERVER_PUSH）。

HTTP 元数据属性

content_type与accept同样走ScopeState缓存（content_type、accept槽位），避免重复解析。

正文读取：body / stream / json / msgpack

• await request.body()：一次性聚合全部正文为bytes；
• request.stream()：异步生成器，逐块产出字节流。其内部实现了请求体大小上限的双重校验：
  1. 若content-length头声明的大小超过route_handler.request_max_body_size（未配置则为math.inf），在读取任何字节前直接抛RequestEntityTooLarge（413）；
  2. 实际累计读取字节数超过上限时也会抛 413——注释特别说明这一层校验是有意保留的，因为请求解压等中间件可能导致流式正文合法地超过 wire 格式的 content-length；
  • 若收到http.disconnect事件则抛InternalServerException("client disconnected prematurely")；流被消费过后再次调用会抛InternalServerException("stream consumed")。
• await request.json()：读取正文并反序列化为任意 Python 对象，空正文按b"null"处理，支持route_handler.type_decoders注册的自定义解码器；
• await request.msgpack()：MessagePack 反序列化，空正文按b"\xc0"（nil）处理。

json()与msgpack()的结果同样会被缓存（ScopeState.json、ScopeState.msgpack），多次调用只解析一次。

表单解析：await request.form()

form()根据Content-Type分派三种路径（litestar/connection/request.py）：

1. multipart/form-data：从 options 中取出boundary（缺失则抛ValidationException），调用parse_multipart_form流式解析，部件数量上限受app.multipart_form_part_limit约束；
2. application/x-www-form-urlencoded：读取正文后调用parse_url_encoded_form_data解析；
3. 其他类型：返回空结果。

最终统一包装为FormMultiDict（不可变多值字典，见 datastructures/multi_dicts.py）返回，同样支持连接级缓存。

HTTP/2 Server Push：send_push_promise

send_push_promise(path, raise_if_unavailable=False)用于在支持http.response.push扩展的服务器上主动推送资源。其行为分为三档：

• 服务器支持：从当前请求头中挑选SERVER_PUSH_HEADERS（accept、accept-encoding、accept-language、cache-control、user-agent，定义于 litestar/connection/request.py）构造推送给定的请求头，然后发送http.response.push事件；
• 服务器不支持且raise_if_unavailable=True：抛LitestarException；
• 服务器不支持且未设置：发出LitestarWarning，并提示可通过request.supports_push_promise动态检测（未来版本将默认改为抛异常）。

WebSocket：全双工通信的完整封装

WebSocket类在构造时就用receive_wrapper/send_wrapper包装了底层收发函数，用connection_state字段（取值"init" | "connect" | "receive" | "disconnect"）跟踪连接生命周期，并依据状态自动拦截异常行为（litestar/connection/websocket.py）：

• receive_wrapper：根据收到的 ASGI 事件类型更新状态——websocket.connect→"connect"、websocket.receive→"receive"、其余（含 disconnect）→"disconnect"；若状态已是"disconnect"仍尝试接收，直接抛WebSocketDisconnect；
• send_wrapper：发送前校验状态，已断开则抛WebSocketDisconnect。

握手与关闭

• await ws.accept(subprotocols=None, headers=None)：接受连接，可指定子协议与响应头（支持Headers、dict或(bytes, bytes)列表三种形式，统一转换为list[tuple[bytes, bytes]]写入websocket.accept事件）；状态为"init"时会先自动消费 connect 事件；
• await ws.close(code=WS_1000_NORMAL_CLOSURE, reason=None)：发送websocket.close事件，默认状态码为 1000（正常关闭，定义于 status_codes.py）。

数据接收 API

数据发送 API

JSON/MessagePack 的编解码统一经由 serialization/msgspec_hooks.py 暴露的decode_json/encode_json/decode_msgpack/encode_msgpack完成，默认序列化器为default_serializer。

连接层在依赖注入中的应用

Request与WebSocket都是可以直接声明为依赖参数的类型：在路由处理器或依赖函数中把参数类型标注为Request/WebSocket（或其子类），Litestar 的参数解析器（litestar/_kwargs/extractors.py）就会注入对应连接实例。此时可以放心使用request.app、request.state、request.auth、request.user、request.session等能力，与 docs/usage/dependency-injection.rst 描述的整体机制协同工作。

典型实战组合：鉴权 + Session + URL 生成

综合上述 API，一个典型的受保护接口可以这样组织：

from litestar import get from litestar.connection import Request @get("/account", name="account") async def account(request: Request) -> dict: # 依赖 AuthMiddleware 注入的 user / auth user = request.user # 读写 Session（依赖 SessionMiddleware） request.set_session({"last_visited": "/account"}) # 生成指向其他路由的绝对 URL profile_url = request.url_for("account") return {"user": str(user), "url": profile_url}

• 若未安装AuthMiddleware而访问request.user，会立即抛出ImproperlyConfiguredException，提示正确装配方式；
• request.url_for("account")基于base_url拼接，挂载在子应用前缀下也能得到正确的绝对地址。

与测试的对应关系

连接层的核心行为都有单元测试覆盖（tests/unit/test_connection/）：

• test_base.py：验证ASGIConnection各属性与ScopeState缓存联动、auth/user/session的注入读取；
• test_connection_caching.py：验证 URL、query、cookies 等解析结果的连接级缓存行为；
• test_request.py：覆盖正文、表单、JSON/MessagePack、content-length 异常等 HTTP 能力；
• test_websocket.py：覆盖握手、收发各模式、断连异常与迭代器行为。

在编写自定义中间件、自定义 Request 子类（参考 docs/examples/request_data/custom_request.py）或直接处理 WebSocket 时，这三份测试是理解预期行为的最佳范本。

小结

litestar.connection模块通过ASGIConnection→Request/WebSocket的两级继承，把 ASGI 协议细节统一收纳进一套带缓存、带状态机、带异常防护的连接 API：

• ASGIConnection提供跨协议公共能力（URL/头/参数/Cookie/State/Session/auth/user/url_for）；
• Request补全 HTTP 正文、表单、JSON/MessagePack 与 Server Push；
• WebSocket用状态机管理全双工收发的握手、编解码与断连语义。

理解这一层抽象，也就理解了 Litestar 的依赖注入、鉴权、Session 中间件乃至 WebSocket 子应用为什么能如此简洁地共享同一套上下文，是深入框架内部的理想切入点。

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar