Starlette 端点Endpoints指南用 HTTPEndpoint 与 WebSocketEndpoint 构建类视图与 WebSocket 服务【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starletteStarlette 在 starlette/endpoints.py 中提供了HTTPEndpoint与WebSocketEndpoint两个类用于以类视图class-based view的方式组织 HTTP 方法分发与 WebSocket 会话处理。本文将以 docs/endpoints.md 为主线结合源码实现与测试用例完整讲解这两类端点的用法、底层分发机制、数据校验规则与 405 响应行为读完你可以在自己的 Starlette 应用中直接写出类视图风格的 HTTP 接口和带自动数据校验的 WebSocket 服务。一、端点是什么Starlette 的类视图模式传统 ASGI 应用是一个async def app(scope, receive, send)风格的函数而 Starlette 的端点Endpoint则把同一套 ASGI 接口封装成可复用的类HTTPEndpoint按 HTTP 请求方法GET / POST / PUT / DELETE 等自动分发到对应的处理器方法等价于许多 Web 框架中的类视图。WebSocketEndpoint围绕WebSocket实例提供连接、接收、断开三个生命周期钩子并可通过encoding属性对收到的数据做格式校验。从路由源码可以印证类端点与函数端点的差异starlette/routing.py 中Route在初始化时通过inspect.isfunction/inspect.ismethod判断端点类型——函数端点会被包装为request_response(endpoint)而类端点直接作为 ASGI 应用使用self.app endpoint。WebSocketRoute的处理逻辑与此一致见 starlette/routing.py。因此使用端点类时必须把类本身传给路由而不是类的实例。二、HTTPEndpointHTTP 方法分发的类视图2.1 作为独立的 ASGI 应用使用HTTPEndpoint类本身就是可用的 ASGI 应用可以直接交给服务器运行from starlette.responses import PlainTextResponse from starlette.endpoints import HTTPEndpoint class App(HTTPEndpoint): async def get(self, request): return PlainTextResponse(fHello, world!)底层原理是HTTPEndpoint实现了__await__它委托给dispatch()协程见 starlette/endpoints.py并在构造函数中断言scope[type] httpstarlette/endpoints.py保证只能用于 HTTP 请求。2.2 与 Starlette 路由系统集成在标准应用中通常把端点类挂到Route上由Starlette应用实例负责分发from starlette.applications import Starlette from starlette.responses import PlainTextResponse from starlette.endpoints import HTTPEndpoint from starlette.routing import Route class Homepage(HTTPEndpoint): async def get(self, request): return PlainTextResponse(fHello, world!) async def query(self, request): return PlainTextResponse(fHello, query!) class User(HTTPEndpoint): async def get(self, request): username request.path_params[username] return PlainTextResponse(fHello, {username}) routes [ Route(/, Homepage), Route(/{username}, User) ] app Starlette(routesroutes)这里有两个要点必须传类本身Route(/, Homepage)而非Homepage(...)因为Route需要把类当 ASGI 应用实例化。路径参数自动可用Route(/{username}, User)会把匹配到的username放进request.path_params处理器方法中直接通过request.path_params[username]读取与 tests/test_endpoints.py 中Homepage的用法一致。2.3 方法分发机制剖析dispatch()的实现starlette/endpoints.py揭示了完整的分发流程async def dispatch(self) - None: request Request(self.scope, receiveself.receive) handler_name get if request.method HEAD and not hasattr(self, head) else request.method.lower() handler: Callable[[Request], Any] if request.method in self._allowed_methods or (request.method HEAD and GET in self._allowed_methods): handler getattr(self, handler_name) else: handler self.method_not_allowed is_async is_async_callable(handler) if is_async: response await handler(request) else: response await run_in_threadpool(handler, request) await response(self.scope, self.receive, self.send)从中可以得到几条对实践很有价值的事实允许的方法列表是动态推导的构造函数中会根据类上是否定义了对应小写方法从(GET, HEAD, POST, PUT, PATCH, DELETE, OPTIONS, QUERY)中筛选出_allowed_methods见 starlette/endpoints.py。也就是说定义了get和query两个方法的类其允许列表就是GET, QUERY。HEAD 自动回退到 GET当请求方法是HEAD且类上没有显式head方法时会直接调用get处理器条件request.method HEAD and GET in self._allowed_methods保证了这一回退的合法性。支持同步处理器处理器不要求必须是async def。dispatch()用is_async_callable判断若为同步函数则通过run_in_threadpool放入线程池执行线程池实现见 starlette/concurrency.py避免阻塞事件循环——这是比普通函数视图更宽容的写法。响应对象本身就是 ASGI 应用最后一行await response(self.scope, self.receive, self.send)说明处理器返回的Response会被当作 ASGI 应用直接调用这与 starlette/responses.py 中Response.__call__的设计一脉相承。2.4 405 响应与 Allow 头原文档指出凡是无法映射到对应处理器方法的请求方法端点类都会返回 405 Method not allowed 响应。源码实现位于method_not_allowedstarlette/endpoints.pyasync def method_not_allowed(self, request: Request) - Response: headers {Allow: , .join(self._allowed_methods)} if app in self.scope: raise HTTPException(status_code405, headersheaders) return PlainTextResponse(Method Not Allowed, status_code405, headersheaders)关键细节是响应方式因运行环境而异若端点在 Starlette 应用内运行scope中存在app键会抛出HTTPException(405)交由应用配置的异常处理器统一处理——这意味着你可以为 405 定制响应内容异常机制见 starlette/exceptions.py。若作为独立 ASGI 应用运行则直接返回PlainTextResponse(Method Not Allowed, status_code405)。两种情况下响应头都会带上Allow字段如实列出该端点支持的方法。测试用例验证了这一点tests/test_endpoints.py对只定义了get和query的端点发送POST返回405且response.headers[allow] GET, QUERY。2.5 特殊方法 QUERY 与非动词方法从源码和测试可以看出两个值得注意的行为支持非标准 HTTP 方法 QUERY_allowed_methods的候选列表中包含QUERY。上面的示例async def query(self, request)就是为 QUERY 请求服务的处理器。tests/test_endpoints.py 用client.request(QUERY, /)验证了该方法会正常返回 200。不以下划线开头的无关方法也不会被分发分发只针对候选方法名其他同名方法会被忽略。tests/test_endpoints.py 中定义了_do_delete这类特权辅助方法对_DO_DELETE请求依然返回 405 且Allow: GET说明端点类不会把内部辅助方法意外暴露为路由端点。三、WebSocketEndpointWebSocket 生命周期封装WebSocketEndpoint是围绕WebSocket实例的 ASGI 应用包装。它通过类属性encoding声明期望的数据格式并提供三个可覆写的生命周期钩子async def on_connect(websocket, **kwargs)处理连接建立async def on_receive(websocket, data)处理收到的数据async def on_disconnect(websocket, close_code)处理连接断开3.1 基本用法from starlette.endpoints import WebSocketEndpoint class App(WebSocketEndpoint): encoding bytes async def on_connect(self, websocket): await websocket.accept() async def on_receive(self, websocket, data): await websocket.send_bytes(bMessage: data) async def on_disconnect(self, websocket, close_code): passencoding支持三种取值用于在on_receive之前校验 WebSocket 数据格式encoding期望的消息内容收到的data类型json合法的 JSON 数据解析后的 Python 对象json.loads结果bytes二进制帧bytestext文本帧strNone默认不校验按实际帧类型透传text或bytes注意encoding是一个类属性starlette/endpoints.py类型标注为Literal[text, bytes, json] | None默认None表示不做格式约束。3.2 默认钩子行为若没有覆写基类提供安全默认值starlette/endpoints.pyon_connect默认调用await websocket.accept()接受连接on_receive默认什么都不做on_disconnect默认什么都不做。也就是说一个最简单的回显端点只需要覆写on_receive即可。3.3 encoding 数据校验与 decode 实现校验逻辑集中在decode()方法starlette/endpoints.py它在on_receive之前执行async def decode(self, websocket: WebSocket, message: Message) - Any: if self.encoding text: if text not in message: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Expected text websocket messages, but got bytes) return message[text] elif self.encoding bytes: if bytes not in message: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Expected bytes websocket messages, but got text) return message[bytes] elif self.encoding json: if message.get(text) is not None: text message[text] else: text message[bytes].decode(utf-8) try: return json.loads(text) except json.decoder.JSONDecodeError: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Malformed JSON data received.) assert self.encoding is None, fUnsupported encoding attribute {self.encoding} return message[text] if message.get(text) else message[bytes]这条实现揭示了几个重要的实战细节格式不匹配时主动断开连接文本期望收到二进制帧、二进制期望收到文本帧、JSON 解析失败三种情况都会先以WS_1003_UNSUPPORTED_DATA值为 1003定义于 starlette/status.py关闭连接再抛出RuntimeError。JSON 兼容两种帧encoding json时无论消息以文本帧message[text]还是二进制帧message[bytes].decode(utf-8)到达都会被解析为 JSON 对象。测试 tests/test_endpoints.py 专门验证了modebinary的 JSON 收发路径。未知 encoding 会触发断言assert self.encoding is None兜底拦截了拼写错误的 encoding 值避免静默错误。3.4 dispatch 生命周期循环WebSocketEndpoint.dispatch()starlette/endpoints.py实现了完整的会话循环async def dispatch(self) - None: websocket WebSocket(self.scope, receiveself.receive, sendself.send) await self.on_connect(websocket) close_code status.WS_1000_NORMAL_CLOSURE try: while True: message await websocket.receive() if message[type] websocket.receive: data await self.decode(websocket, message) await self.on_receive(websocket, data) elif message[type] websocket.disconnect: # pragma: no branch close_code int(message.get(code) or status.WS_1000_NORMAL_CLOSURE) break except Exception as exc: close_code status.WS_1011_INTERNAL_ERROR raise exc finally: await self.on_disconnect(websocket, close_code)生命周期一目了然创建WebSocket包装对象调用on_connect默认接受连接进入消息循环websocket.receive()会校验 ASGI 状态转换见 starlette/websockets.py收到websocket.receive消息 →decode校验 →on_receive收到websocket.disconnect→ 记录对端关闭码默认WS_1000_NORMAL_CLOSURE即 1000并退出循环若处理过程中抛出异常关闭码改为WS_1011_INTERNAL_ERROR1001 之外的内部错误码见 starlette/status.py并向上抛出无论正常还是异常退出finally中都会调用on_disconnect(websocket, close_code)把真实的关闭码交给子类。这意味着on_disconnect总能拿到会话的最终关闭码。测试 tests/test_endpoints.py 验证了客户端以1001WS_1001_GOING_AWAY主动关闭时on_disconnect收到的close_code 1001。3.5 与 Starlette 应用配合完整聊天室示例WebSocketEndpoint可以像HTTPEndpoint一样挂载到WebSocketRoute上与原文档中的完整示例一致——用HTTPEndpoint提供聊天室页面用WebSocketEndpoint处理实时回显import uvicorn from starlette.applications import Starlette from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint from starlette.responses import HTMLResponse from starlette.routing import Route, WebSocketRoute html !DOCTYPE html html head titleChat/title /head body h1WebSocket Chat/h1 form action onsubmitsendMessage(event) input typetext idmessageText autocompleteoff/ buttonSend/button /form ul idmessages /ul script var ws new WebSocket(ws://localhost:8000/ws); ws.onmessage function(event) { var messages document.getElementById(messages) var message document.createElement(li) var content document.createTextNode(event.data) message.appendChild(content) messages.appendChild(message) }; function sendMessage(event) { var input document.getElementById(messageText) ws.send(input.value) input.value event.preventDefault() } /script /body /html class Homepage(HTTPEndpoint): async def get(self, request): return HTMLResponse(html) class Echo(WebSocketEndpoint): encoding text async def on_receive(self, websocket, data): await websocket.send_text(fMessage text was: {data}) routes [ Route(/, Homepage), WebSocketRoute(/ws, Echo) ] app Starlette(routesroutes)运行uvicorn app:app假定上述代码保存在app.py后浏览器访问http://localhost:8000/即可打开聊天页面向ws://localhost:8000/ws发送的每条文本都会被服务端回显。由于Echo.encoding text二进制帧会被自动以WS_1003关闭并抛出RuntimeError——这正是 tests/test_endpoints.py 所验证的行为。四、测试验证与实战建议4.1 测试用例对照端点行为在 tests/test_endpoints.py 中有完整覆盖可作为学习与回归参考测试函数验证点test_http_endpoint_route/test_http_endpoint_route_path_params基础 GET 分发与路径参数注入test_http_endpoint_route_method未定义方法返回 405 且带Allow头test_http_endpoint_route_query_method非标准 QUERY 方法可正常分发test_http_endpoint_does_not_dispatch_non_verb_method内部辅助方法_do_delete不会被暴露test_websocket_endpoint_on_connecton_connect中可校验并选择 subprotocoltest_websocket_endpoint_on_receive_bytes/_json/_json_binary/_text四种 encoding 的收发与格式校验不匹配时抛RuntimeErrortest_websocket_endpoint_on_defaultencoding None时不校验、透传文本帧test_websocket_endpoint_on_disconnecton_disconnect收到真实关闭码4.2 实战建议选择函数视图还是端点类端点类适合把一组相关方法CRUD、聊天协议聚合在一个类里配合request.path_params复用路由逻辑单方法接口用函数视图更简洁。二者可混用因为 starlette/routing.py 对函数和类端点都做了兼容。让框架接管 405 的呈现把端点挂到Starlette应用上405 会以HTTPException抛出从而可以被全局异常处理器或中间件统一美化独立使用端点类时则直接得到PlainTextResponse。WebSocket 务必声明 encoding设置encoding后decode()会在进入on_receive前完成格式校验把脏数据挡在业务逻辑之外但要注意格式错误会以WS_1003关闭连接客户端需要能处理该关闭码。善用钩子做资源清理on_disconnect的finally语义保证总会执行适合做连接计数、日志记录、状态清理等收尾工作。进一步阅读docs/endpoints.md本指南原文、docs/routing.md路由与 URL 转换器、docs/websockets.mdWebSocket实例的完整 API包括accept/send_text/send_bytes/send_json/receive_json等见 starlette/websockets.py、starlette/status.pyWebSocket 关闭码常量定义。【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Starlette 端点（Endpoints）指南：用 HTTPEndpoint 与 WebSocketEndpoint 构建类视图与 WebSocket 服务
Starlette 端点Endpoints指南用 HTTPEndpoint 与 WebSocketEndpoint 构建类视图与 WebSocket 服务【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starletteStarlette 在 starlette/endpoints.py 中提供了HTTPEndpoint与WebSocketEndpoint两个类用于以类视图class-based view的方式组织 HTTP 方法分发与 WebSocket 会话处理。本文将以 docs/endpoints.md 为主线结合源码实现与测试用例完整讲解这两类端点的用法、底层分发机制、数据校验规则与 405 响应行为读完你可以在自己的 Starlette 应用中直接写出类视图风格的 HTTP 接口和带自动数据校验的 WebSocket 服务。一、端点是什么Starlette 的类视图模式传统 ASGI 应用是一个async def app(scope, receive, send)风格的函数而 Starlette 的端点Endpoint则把同一套 ASGI 接口封装成可复用的类HTTPEndpoint按 HTTP 请求方法GET / POST / PUT / DELETE 等自动分发到对应的处理器方法等价于许多 Web 框架中的类视图。WebSocketEndpoint围绕WebSocket实例提供连接、接收、断开三个生命周期钩子并可通过encoding属性对收到的数据做格式校验。从路由源码可以印证类端点与函数端点的差异starlette/routing.py 中Route在初始化时通过inspect.isfunction/inspect.ismethod判断端点类型——函数端点会被包装为request_response(endpoint)而类端点直接作为 ASGI 应用使用self.app endpoint。WebSocketRoute的处理逻辑与此一致见 starlette/routing.py。因此使用端点类时必须把类本身传给路由而不是类的实例。二、HTTPEndpointHTTP 方法分发的类视图2.1 作为独立的 ASGI 应用使用HTTPEndpoint类本身就是可用的 ASGI 应用可以直接交给服务器运行from starlette.responses import PlainTextResponse from starlette.endpoints import HTTPEndpoint class App(HTTPEndpoint): async def get(self, request): return PlainTextResponse(fHello, world!)底层原理是HTTPEndpoint实现了__await__它委托给dispatch()协程见 starlette/endpoints.py并在构造函数中断言scope[type] httpstarlette/endpoints.py保证只能用于 HTTP 请求。2.2 与 Starlette 路由系统集成在标准应用中通常把端点类挂到Route上由Starlette应用实例负责分发from starlette.applications import Starlette from starlette.responses import PlainTextResponse from starlette.endpoints import HTTPEndpoint from starlette.routing import Route class Homepage(HTTPEndpoint): async def get(self, request): return PlainTextResponse(fHello, world!) async def query(self, request): return PlainTextResponse(fHello, query!) class User(HTTPEndpoint): async def get(self, request): username request.path_params[username] return PlainTextResponse(fHello, {username}) routes [ Route(/, Homepage), Route(/{username}, User) ] app Starlette(routesroutes)这里有两个要点必须传类本身Route(/, Homepage)而非Homepage(...)因为Route需要把类当 ASGI 应用实例化。路径参数自动可用Route(/{username}, User)会把匹配到的username放进request.path_params处理器方法中直接通过request.path_params[username]读取与 tests/test_endpoints.py 中Homepage的用法一致。2.3 方法分发机制剖析dispatch()的实现starlette/endpoints.py揭示了完整的分发流程async def dispatch(self) - None: request Request(self.scope, receiveself.receive) handler_name get if request.method HEAD and not hasattr(self, head) else request.method.lower() handler: Callable[[Request], Any] if request.method in self._allowed_methods or (request.method HEAD and GET in self._allowed_methods): handler getattr(self, handler_name) else: handler self.method_not_allowed is_async is_async_callable(handler) if is_async: response await handler(request) else: response await run_in_threadpool(handler, request) await response(self.scope, self.receive, self.send)从中可以得到几条对实践很有价值的事实允许的方法列表是动态推导的构造函数中会根据类上是否定义了对应小写方法从(GET, HEAD, POST, PUT, PATCH, DELETE, OPTIONS, QUERY)中筛选出_allowed_methods见 starlette/endpoints.py。也就是说定义了get和query两个方法的类其允许列表就是GET, QUERY。HEAD 自动回退到 GET当请求方法是HEAD且类上没有显式head方法时会直接调用get处理器条件request.method HEAD and GET in self._allowed_methods保证了这一回退的合法性。支持同步处理器处理器不要求必须是async def。dispatch()用is_async_callable判断若为同步函数则通过run_in_threadpool放入线程池执行线程池实现见 starlette/concurrency.py避免阻塞事件循环——这是比普通函数视图更宽容的写法。响应对象本身就是 ASGI 应用最后一行await response(self.scope, self.receive, self.send)说明处理器返回的Response会被当作 ASGI 应用直接调用这与 starlette/responses.py 中Response.__call__的设计一脉相承。2.4 405 响应与 Allow 头原文档指出凡是无法映射到对应处理器方法的请求方法端点类都会返回 405 Method not allowed 响应。源码实现位于method_not_allowedstarlette/endpoints.pyasync def method_not_allowed(self, request: Request) - Response: headers {Allow: , .join(self._allowed_methods)} if app in self.scope: raise HTTPException(status_code405, headersheaders) return PlainTextResponse(Method Not Allowed, status_code405, headersheaders)关键细节是响应方式因运行环境而异若端点在 Starlette 应用内运行scope中存在app键会抛出HTTPException(405)交由应用配置的异常处理器统一处理——这意味着你可以为 405 定制响应内容异常机制见 starlette/exceptions.py。若作为独立 ASGI 应用运行则直接返回PlainTextResponse(Method Not Allowed, status_code405)。两种情况下响应头都会带上Allow字段如实列出该端点支持的方法。测试用例验证了这一点tests/test_endpoints.py对只定义了get和query的端点发送POST返回405且response.headers[allow] GET, QUERY。2.5 特殊方法 QUERY 与非动词方法从源码和测试可以看出两个值得注意的行为支持非标准 HTTP 方法 QUERY_allowed_methods的候选列表中包含QUERY。上面的示例async def query(self, request)就是为 QUERY 请求服务的处理器。tests/test_endpoints.py 用client.request(QUERY, /)验证了该方法会正常返回 200。不以下划线开头的无关方法也不会被分发分发只针对候选方法名其他同名方法会被忽略。tests/test_endpoints.py 中定义了_do_delete这类特权辅助方法对_DO_DELETE请求依然返回 405 且Allow: GET说明端点类不会把内部辅助方法意外暴露为路由端点。三、WebSocketEndpointWebSocket 生命周期封装WebSocketEndpoint是围绕WebSocket实例的 ASGI 应用包装。它通过类属性encoding声明期望的数据格式并提供三个可覆写的生命周期钩子async def on_connect(websocket, **kwargs)处理连接建立async def on_receive(websocket, data)处理收到的数据async def on_disconnect(websocket, close_code)处理连接断开3.1 基本用法from starlette.endpoints import WebSocketEndpoint class App(WebSocketEndpoint): encoding bytes async def on_connect(self, websocket): await websocket.accept() async def on_receive(self, websocket, data): await websocket.send_bytes(bMessage: data) async def on_disconnect(self, websocket, close_code): passencoding支持三种取值用于在on_receive之前校验 WebSocket 数据格式encoding期望的消息内容收到的data类型json合法的 JSON 数据解析后的 Python 对象json.loads结果bytes二进制帧bytestext文本帧strNone默认不校验按实际帧类型透传text或bytes注意encoding是一个类属性starlette/endpoints.py类型标注为Literal[text, bytes, json] | None默认None表示不做格式约束。3.2 默认钩子行为若没有覆写基类提供安全默认值starlette/endpoints.pyon_connect默认调用await websocket.accept()接受连接on_receive默认什么都不做on_disconnect默认什么都不做。也就是说一个最简单的回显端点只需要覆写on_receive即可。3.3 encoding 数据校验与 decode 实现校验逻辑集中在decode()方法starlette/endpoints.py它在on_receive之前执行async def decode(self, websocket: WebSocket, message: Message) - Any: if self.encoding text: if text not in message: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Expected text websocket messages, but got bytes) return message[text] elif self.encoding bytes: if bytes not in message: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Expected bytes websocket messages, but got text) return message[bytes] elif self.encoding json: if message.get(text) is not None: text message[text] else: text message[bytes].decode(utf-8) try: return json.loads(text) except json.decoder.JSONDecodeError: await websocket.close(codestatus.WS_1003_UNSUPPORTED_DATA) raise RuntimeError(Malformed JSON data received.) assert self.encoding is None, fUnsupported encoding attribute {self.encoding} return message[text] if message.get(text) else message[bytes]这条实现揭示了几个重要的实战细节格式不匹配时主动断开连接文本期望收到二进制帧、二进制期望收到文本帧、JSON 解析失败三种情况都会先以WS_1003_UNSUPPORTED_DATA值为 1003定义于 starlette/status.py关闭连接再抛出RuntimeError。JSON 兼容两种帧encoding json时无论消息以文本帧message[text]还是二进制帧message[bytes].decode(utf-8)到达都会被解析为 JSON 对象。测试 tests/test_endpoints.py 专门验证了modebinary的 JSON 收发路径。未知 encoding 会触发断言assert self.encoding is None兜底拦截了拼写错误的 encoding 值避免静默错误。3.4 dispatch 生命周期循环WebSocketEndpoint.dispatch()starlette/endpoints.py实现了完整的会话循环async def dispatch(self) - None: websocket WebSocket(self.scope, receiveself.receive, sendself.send) await self.on_connect(websocket) close_code status.WS_1000_NORMAL_CLOSURE try: while True: message await websocket.receive() if message[type] websocket.receive: data await self.decode(websocket, message) await self.on_receive(websocket, data) elif message[type] websocket.disconnect: # pragma: no branch close_code int(message.get(code) or status.WS_1000_NORMAL_CLOSURE) break except Exception as exc: close_code status.WS_1011_INTERNAL_ERROR raise exc finally: await self.on_disconnect(websocket, close_code)生命周期一目了然创建WebSocket包装对象调用on_connect默认接受连接进入消息循环websocket.receive()会校验 ASGI 状态转换见 starlette/websockets.py收到websocket.receive消息 →decode校验 →on_receive收到websocket.disconnect→ 记录对端关闭码默认WS_1000_NORMAL_CLOSURE即 1000并退出循环若处理过程中抛出异常关闭码改为WS_1011_INTERNAL_ERROR1001 之外的内部错误码见 starlette/status.py并向上抛出无论正常还是异常退出finally中都会调用on_disconnect(websocket, close_code)把真实的关闭码交给子类。这意味着on_disconnect总能拿到会话的最终关闭码。测试 tests/test_endpoints.py 验证了客户端以1001WS_1001_GOING_AWAY主动关闭时on_disconnect收到的close_code 1001。3.5 与 Starlette 应用配合完整聊天室示例WebSocketEndpoint可以像HTTPEndpoint一样挂载到WebSocketRoute上与原文档中的完整示例一致——用HTTPEndpoint提供聊天室页面用WebSocketEndpoint处理实时回显import uvicorn from starlette.applications import Starlette from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint from starlette.responses import HTMLResponse from starlette.routing import Route, WebSocketRoute html !DOCTYPE html html head titleChat/title /head body h1WebSocket Chat/h1 form action onsubmitsendMessage(event) input typetext idmessageText autocompleteoff/ buttonSend/button /form ul idmessages /ul script var ws new WebSocket(ws://localhost:8000/ws); ws.onmessage function(event) { var messages document.getElementById(messages) var message document.createElement(li) var content document.createTextNode(event.data) message.appendChild(content) messages.appendChild(message) }; function sendMessage(event) { var input document.getElementById(messageText) ws.send(input.value) input.value event.preventDefault() } /script /body /html class Homepage(HTTPEndpoint): async def get(self, request): return HTMLResponse(html) class Echo(WebSocketEndpoint): encoding text async def on_receive(self, websocket, data): await websocket.send_text(fMessage text was: {data}) routes [ Route(/, Homepage), WebSocketRoute(/ws, Echo) ] app Starlette(routesroutes)运行uvicorn app:app假定上述代码保存在app.py后浏览器访问http://localhost:8000/即可打开聊天页面向ws://localhost:8000/ws发送的每条文本都会被服务端回显。由于Echo.encoding text二进制帧会被自动以WS_1003关闭并抛出RuntimeError——这正是 tests/test_endpoints.py 所验证的行为。四、测试验证与实战建议4.1 测试用例对照端点行为在 tests/test_endpoints.py 中有完整覆盖可作为学习与回归参考测试函数验证点test_http_endpoint_route/test_http_endpoint_route_path_params基础 GET 分发与路径参数注入test_http_endpoint_route_method未定义方法返回 405 且带Allow头test_http_endpoint_route_query_method非标准 QUERY 方法可正常分发test_http_endpoint_does_not_dispatch_non_verb_method内部辅助方法_do_delete不会被暴露test_websocket_endpoint_on_connecton_connect中可校验并选择 subprotocoltest_websocket_endpoint_on_receive_bytes/_json/_json_binary/_text四种 encoding 的收发与格式校验不匹配时抛RuntimeErrortest_websocket_endpoint_on_defaultencoding None时不校验、透传文本帧test_websocket_endpoint_on_disconnecton_disconnect收到真实关闭码4.2 实战建议选择函数视图还是端点类端点类适合把一组相关方法CRUD、聊天协议聚合在一个类里配合request.path_params复用路由逻辑单方法接口用函数视图更简洁。二者可混用因为 starlette/routing.py 对函数和类端点都做了兼容。让框架接管 405 的呈现把端点挂到Starlette应用上405 会以HTTPException抛出从而可以被全局异常处理器或中间件统一美化独立使用端点类时则直接得到PlainTextResponse。WebSocket 务必声明 encoding设置encoding后decode()会在进入on_receive前完成格式校验把脏数据挡在业务逻辑之外但要注意格式错误会以WS_1003关闭连接客户端需要能处理该关闭码。善用钩子做资源清理on_disconnect的finally语义保证总会执行适合做连接计数、日志记录、状态清理等收尾工作。进一步阅读docs/endpoints.md本指南原文、docs/routing.md路由与 URL 转换器、docs/websockets.mdWebSocket实例的完整 API包括accept/send_text/send_bytes/send_json/receive_json等见 starlette/websockets.py、starlette/status.pyWebSocket 关闭码常量定义。【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考