企业官网建设流程全解析

Starlette 端点（Endpoints）指南：用 HTTPEndpoint 与 WebSocketEndpoint 构建类视图与 WebSocket 服务

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

Starlette 在 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）。因此使用端点类时必须把类本身传给路由，而不是类的实例。

二、HTTPEndpoint：HTTP 方法分发的类视图

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(f"Hello, world!")

底层原理是：HTTPEndpoint实现了__await__，它委托给dispatch()协程（见 starlette/endpoints.py），并在构造函数中断言scope["type"] == "http"（starlette/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(f"Hello, world!") async def query(self, request): return PlainTextResponse(f"Hello, query!") class User(HTTPEndpoint): async def get(self, request): username = request.path_params['username'] return PlainTextResponse(f"Hello, {username}") routes = [ Route("/", Homepage), Route("/{username}", User) ] app = Starlette(routes=routes)

这里有两个要点：

1. 必须传类本身：Route("/", Homepage)而非Homepage(...)，因为Route需要把类当 ASGI 应用实例化。
2. 路径参数自动可用：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, receive=self.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_allowed（starlette/endpoints.py）：

async def method_not_allowed(self, request: Request) -> Response: headers = {"Allow": ", ".join(self._allowed_methods)} if "app" in self.scope: raise HTTPException(status_code=405, headers=headers) return PlainTextResponse("Method Not Allowed", status_code=405, headers=headers)

关键细节是响应方式因运行环境而异：

• 若端点在 Starlette 应用内运行（scope中存在"app"键），会抛出HTTPException(405)，交由应用配置的异常处理器统一处理——这意味着你可以为 405 定制响应内容（异常机制见 starlette/exceptions.py）。
• 若作为独立 ASGI 应用运行，则直接返回PlainTextResponse("Method Not Allowed", status_code=405)。

两种情况下响应头都会带上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，说明端点类不会把内部辅助方法意外暴露为路由端点。

三、WebSocketEndpoint：WebSocket 生命周期封装

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(b"Message: " + data) async def on_disconnect(self, websocket, close_code): pass

encoding支持三种取值，用于在on_receive之前校验 WebSocket 数据格式：

注意：encoding是一个类属性（starlette/endpoints.py），类型标注为Literal["text", "bytes", "json"] | None，默认None表示不做格式约束。

3.2 默认钩子行为

若没有覆写，基类提供安全默认值（starlette/endpoints.py）：

• on_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(code=status.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(code=status.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(code=status.WS_1003_UNSUPPORTED_DATA) raise RuntimeError("Malformed JSON data received.") assert self.encoding is None, f"Unsupported '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 专门验证了mode="binary"的 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, receive=self.receive, send=self.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)

生命周期一目了然：

1. 创建WebSocket包装对象，调用on_connect（默认接受连接）；
2. 进入消息循环，websocket.receive()会校验 ASGI 状态转换（见 starlette/websockets.py）；
3. 收到websocket.receive消息 →decode校验 →on_receive；
4. 收到websocket.disconnect→ 记录对端关闭码（默认WS_1000_NORMAL_CLOSURE，即 1000）并退出循环；
5. 若处理过程中抛出异常，关闭码改为WS_1011_INTERNAL_ERROR（1001 之外的内部错误码，见 starlette/status.py）并向上抛出；
6. 无论正常还是异常退出，finally中都会调用on_disconnect(websocket, close_code)，把真实的关闭码交给子类。

这意味着on_disconnect总能拿到会话的最终关闭码。测试 tests/test_endpoints.py 验证了客户端以1001（WS_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> <title>Chat</title> </head> <body> <h1>WebSocket Chat</h1> <form action="" onsubmit="sendMessage(event)"> <input type="text" id="messageText" autocomplete="off"/> <button>Send</button> </form> <ul id='messages'> </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(f"Message text was: {data}") routes = [ Route("/", Homepage), WebSocketRoute("/ws", Echo) ] app = Starlette(routes=routes)

运行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 中有完整覆盖，可作为学习与回归参考：

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.md（WebSocket实例的完整 API，包括accept/send_text/send_bytes/send_json/receive_json等，见 starlette/websockets.py）、starlette/status.py（WebSocket 关闭码常量定义）。

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