尧图网页设计 
FastAPI WebSockets 实战指南连接建立、消息收发、依赖注入与多客户端广播【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiWebSocket 在 HTTP 之上提供一条全双工的持久化通道非常适合聊天、实时通知、协同编辑、进度推送等低延迟双向交互场景。FastAPI在其高级用法中原生支持 WebSocket你可以用装饰器直接声明 WebSocket 路由在异步端点里await收发文本、二进制与 JSON 数据还能像普通 HTTP 路径操作一样使用Depends、Security、Cookie、Header、Path、Query做参数解析与依赖注入。读完本文你将掌握在 FastAPI 中接入websockets依赖、创建并调试最小可用的 WebSocket 服务、在端点内使用依赖校验身份以及用连接管理器处理断连并向多客户端广播消息的完整方案。本文以仓库中的 WebSockets 官方文档法语版与 英文原版 内容一致为主线并结合 fastapi 源码与 docs_src/websockets_ 下的可运行教程示例展开。所有代码示例均取自教程源码可直接复制运行。WebSocket 与 FastAPI从 HTTP 请求-响应到全双工长连接普通 HTTP 接口遵循请求-响应模型客户端发起请求服务端返回响应后连接即告一段落服务端无法主动向客户端推送数据。而 WebSocket 协议在建立连接后服务端与客户端可以在同一条持久连接上随时互相发送消息。在 FastAPI 中启用 WebSocket 非常简单声明一个以app.websocket()装饰的路由路由函数接收一个WebSocket类型的参数即可在其上await接收消息、主动发送消息。FastAPI 并没有另起炉灶实现一套协议栈而是直接复用了 Starlette 的 WebSocket 实现。查看 fastapi/websockets.py 可以看到这个模块只有三行将 Starlette 的WebSocket、WebSocketDisconnect、WebSocketState原样重新导出from starlette.websockets import WebSocket as WebSocket # noqa from starlette.websockets import WebSocketDisconnect as WebSocketDisconnect # noqa from starlette.websockets import WebSocketState as WebSocketState # noqa而在 fastapi/init.py 中WebSocket、WebSocketDisconnect、WebSocketException都通过这个模块对外暴露。因此你可以直接from fastapi import WebSocket而无需关心它来自 Starlette——FastAPI 只是帮你省去了这一层导入细节。第一步安装websockets协议库WebSocket 的协议编解码需要底层库支持。在 FastAPI 中使用 WebSocket 前先把 Python 的websockets库加入项目依赖$ uv add websockets --- 100%websockets是一个以纯 Python 实现的 WebSocket 协议库负责完成握手、帧解析、消息编解码等底层工作。本文所在的仓库是一个 FastAPI 文档与代码库其依赖锁定文件 uv.lock 使用uv管理因此官方教程统一采用uv add命令添加依赖。如果你使用其他包管理器也可等价地执行pip install websockets等命令效果相同。建立一个最小 WebSocket 通道要演示 WebSocket 的服务端行为需要一个客户端。官方文档指出在真实生产环境中客户端通常是基于 React、Vue.js 或 Angular 等现代框架构建的前端直接调用其 WebSocket 工具直接与后端 WebSocket 端点通信的原生移动应用其他任何能够访问该 WebSocket 端点的客户端程序。但为了把注意力集中在服务端逻辑上教程采用了最简单的方式——把一段内嵌 JavaScript 的 HTML 放在一个超长字符串里作为演示客户端页面。这种方式不适合生产但足够让你在浏览器里立刻体验连接与收发消息的完整链路。完整示例见 docs_src/websockets_/tutorial001_py310.pyfrom fastapi import FastAPI, WebSocket from fastapi.responses import HTMLResponse app FastAPI() html ...一段内嵌 JavaScript 的 HTML 页面JavaScript 中创建 new WebSocket(ws://localhost:8000/ws) 并处理 onmessage / send app.get(/) async def get(): return HTMLResponse(html) app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text(fMessage text was: {data})其中 HTML 字符串第 6–38 行里最关键的是浏览器端逻辑var ws new WebSocket(ws://localhost:8000/ws); ws.onmessage function(event) { // 把服务端发回的消息追加到 ul idmessages 列表中 }; function sendMessage(event) { var input document.getElementById(messageText); ws.send(input.value); // 向服务端发送消息 input.value ; event.preventDefault(); }ws://localhost:8000/ws中的地址必须与你定义的 WebSocket 路由路径一致且当页面通过 HTTP 提供服务时WebSocket 也需使用对应的ws://HTTPS 下则为wss://。路由装饰器与端点签名创建 WebSocket 端点只需要两处关键代码app.websocket(/ws) # ① 声明 WebSocket 路由 async def websocket_endpoint(websocket: WebSocket): # ② 参数标注为 WebSocketapp.websocket(/ws)与app.get(/...)的使用方式一致只是走 WebSocket 协议端点函数第一个参数websocket: WebSocket就是当前连接对象。这里WebSocket由fastapi直接导出等价于from starlette.websockets import WebSocket二者是同一个类。等待与发送消息文本、二进制与 JSON进入端点函数后WebSocket 的操作是一个事件循环。教程中的核心收发逻辑为await websocket.accept() # ① 接受连接完成握手 while True: # ② 持续监听 data await websocket.receive_text() # ③ 等待客户端消息 await websocket.send_text(fMessage text was: {data}) # ④ 回发消息四个步骤各有讲究await websocket.accept()必须显式接受连接。只有在调用之后浏览器端onopen才会触发双方才能开始交换数据。while True长循环使协程持续挂起在接收操作上直到连接关闭抛异常为止。await websocket.receive_text()挂起等待并返回下一条文本消息。若客户端断开这里会抛出WebSocketDisconnect详见下文断连处理一节。await websocket.send_text(...)向客户端发送一条文本消息。除文本外同一套WebSocket对象还支持二进制数据与 JSON你可以用相应的 receive/send 系列方法收发原始字节或直接收发可被 JSON 序列化的结构化数据。也就是说文本、二进制、JSON 三类载荷都可以在这条连接上双向传输具体使用哪一组 API 取决于业务数据的形态。把教程跑起来启动服务并在浏览器中验证将完整示例代码写入main.py然后启动开发服务器$ uv run fastapi dev INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)在浏览器中打开 http://127.0.0.1:8000即可看到教程中的演示页面页面顶部是 WebSocket Chat 标题下方有一个消息输入框和 Send 按钮。在输入框中输入文字并点击发送或回车浏览器端 JavaScript 会把它经ws://localhost:8000/ws发给服务端服务端收到后回发Message text was: 你的输入ws.onmessage再把回显追加到页面的消息列表里。你可以连续发送多条消息它们全部复用同一条WebSocket 连接——这正是 WebSocket 与每次请求新建连接的 HTTP 之间的本质区别。在 WebSocket 端点中使用Depends与其他工具WebSocket 端点并不局限在收发消息上。在app.websocket端点内部你同样可以从fastapi导入并使用这些对象用法与普通路径操作完全一致Depends—— 声明并执行依赖Security—— 带安全语义的依赖Cookie—— 读取请求 CookieHeader—— 读取请求头Path—— 声明路径参数Query—— 声明查询参数教程的第二个示例 docs_src/websockets_/tutorial002_an_py310.py 演示了把认证逻辑放进依赖的做法仓库同时提供了对应的非Annotated写法 tutorial002_py310.py。先看依赖函数它同时声明了一个可选 Cookie 和一个可选查询参数async def get_cookie_or_token( websocket: WebSocket, session: Annotated[str | None, Cookie()] None, token: Annotated[str | None, Query()] None, ): if session is None and token is None: raise WebSocketException(codestatus.WS_1008_POLICY_VIOLATION) return session or token再看 WebSocket 端点它把依赖结果注入进来app.websocket(/items/{item_id}/ws) async def websocket_endpoint( *, websocket: WebSocket, item_id: str, # 来自 URL 路径 q: int | None None, # 来自查询串 cookie_or_token: Annotated[str, Depends(get_cookie_or_token)], # 依赖注入 ): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text( fSession cookie or query token value is: {cookie_or_token} ) if q is not None: await websocket.send_text(fQuery parameter q is: {q}) await websocket.send_text(fMessage text was: {data}, for item ID: {item_id})这段代码展示了三个要点路径参数照常可用item_id从/items/{item_id}/ws中解析查询参数照常可用q可直接作为函数参数声明q: int | None None也可像token那样放进依赖依赖的返回值随消息循环使用cookie_or_token在连接期间保持一致端点每收到一条消息都会引用它。注意在 WebSocket 端点中依赖的声明顺序没有强制要求websocket对象只需在参数列表中声明即可被自动注入。上述代码把websocket放进了仅限关键字的参数区*之后这只是一种风格选择。为什么这里要抛WebSocketException而不是HTTPException依赖校验失败时代码抛出的是WebSocketException(codestatus.WS_1008_POLICY_VIOLATION)。官方文档特别强调既然是 WebSocket 通道再抛HTTPException它是为 HTTP 响应设计的并没有意义正确的做法是抛出WebSocketException并使用 RFC 6455 规范 第 7.4.1 节中定义的合法关闭码——status.WS_1008_POLICY_VIOLATION1008策略违规就是其中之一。仓库源码印证了这一设计。在 fastapi/exceptions.py 中WebSocketException继承自 Starlette 的WebSocketException其构造参数是code来自规范合法关闭码表的一个整数reason可选关闭原因字符串它是 UTF-8 编码数据具体含义由应用自行解释规范并不规定其内容。同时 fastapi/init.py 将它随包导出因此示例里可以直接from fastapi import WebSocketException, status。在依赖校验场景中codestatus.WS_1008_POLICY_VIOLATION会告诉客户端本次关闭是因为策略认证不通过客户端可据此区分关闭类型。用带依赖的示例实测同样把代码放入main.py并启动uv run fastapi dev打开 http://127.0.0.1:8000页面会变成第二个示例的样子——需要先填写两个字段Item ID会拼进 URL 路径例如foo对应/items/foo/wsToken作为查询参数随连接请求发送例如some-key-token它将被依赖函数get_cookie_or_token捕获。点击 Connect 建立连接后再通过 Message 输入框发送消息。由于没有提供名为session的 Cookieget_cookie_or_token会取查询参数token作为凭据每次消息回显的第一行都会报告Session cookie or query token value is: some-key-token若同时传了查询参数q还会追加一行Query parameter q is: ...最后回显原始消息与item_id。如果把 Cookie 与 Token 都留空依赖会立即抛出WebSocketException连接将被以 1008 关闭码拒绝。处理断连与多客户端广播单条连接的收发很简单但真实应用如聊天室往往需要同时维护多条连接并向所有人广播。第三个示例 docs_src/websockets_/tutorial003_py310.py 展示了一个用内存列表管理的ConnectionManagerclass ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() # 接受连接 self.active_connections.append(websocket) # 登记连接 def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) # 移除连接 async def send_personal_message(self, message: str, websocket: WebSocket): await websocket.send_text(message) # 仅发给某个客户端 async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) # 广播给所有客户端 manager ConnectionManager()关键异常处理逻辑在端点里app.websocket(/ws/{client_id}) async def websocket_endpoint(websocket: WebSocket, client_id: int): await manager.connect(websocket) try: while True: data await websocket.receive_text() await manager.send_personal_message(fYou wrote: {data}, websocket) await manager.broadcast(fClient #{client_id} says: {data}) except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast(fClient #{client_id} left the chat)工作流程可以拆成三段登记manager.connect()接受连接并把WebSocket加入内存列表同时给该连接分配一个client_id浏览器端用Date.now()生成并显示在页面标题旁。消息循环每次收到一条文本先send_personal_message把你说了什么单独回给发送者再broadcast把某号客户端说了什么通知到列表里的所有人。异常退出某个客户端关闭页面/标签页时连接断开此时await websocket.receive_text()会抛出WebSocketDisconnect异常——它同样来自 Starlette 并通过 fastapi/websockets.py 导出可直接from fastapi import WebSocketDisconnect。except捕获后调用manager.disconnect()把它从列表移除并广播一条离开消息其余客户端会收到形如Client #1596980209979 left the chat动手试多客户端在多个浏览器标签页中打开应用从各个标签页分别发消息观察所有标签页都能收到彼此的广播然后关闭其中一个标签页——其余客户端会立刻收到某某 left the chat的广播说明断连已被正确侦测并清理。内存实现的边界何时该升级为消息总线官方文档提醒上述ConnectionManager是一个极简演示所有连接都被保存在单个进程的一块内存列表里因此它只在当前进程存活期间有效且只支持单进程部署。一旦你使用多进程如--workers 1或需要跨机器伸缩不同进程各自持有的连接列表互不可见广播就会漏发。如果你的场景需要与 FastAPI 无缝集成但又要求更强的健壮性——例如由 Redis、PostgreSQL 等外部组件做发布订阅支撑的广播方案——官方文档建议参考基于这些消息中间件的广播类库如encode/broadcaster来设计跨进程的消息分发。把连接注册表从进程内迁出、将广播语义交给 pub/sub 中间件是这类架构的通用演进方向。从源码看 FastAPI 如何承载 WebSocket 路由如果你好奇app.websocket()背后的机制可以顺着路由层追踪在 fastapi/routing.py 中定义了APIWebSocketRoute继承自 Starlette 的WebSocketRoute见第 801 行附近它负责把端点函数解析成路由 依赖 参数模型的组合当应用注册app.websocket(/ws)这类路由时约第 3047 行FastAPI 会用APIWebSocketRoute包装你的协程函数并在每个连接进入时完成参数校验与依赖求解——这正是 WebSocket 端点也能使用Depends、Cookie、Query等声明式工具的底层原因。仓库的测试也覆盖了这些能力tests/test_ws_dependencies.py 验证 WebSocket 路由上的依赖注入行为tests/test_ws_router.py 验证路由器上挂载 WebSocket 路由的行为。想确认某一种具体写法Annotated 依赖、路径参数、异常码是否可用最直接的方式就是在这些测试里找到对应的用例或照抄官方教程中的示例代码运行验证。小结本文从零到一地梳理了 FastAPI WebSocket 的完整链路先用uv add websockets引入协议库再通过app.websocket()声明路由、await websocket.accept()完成握手随后在while True循环中receive_text/send_text双向收发消息更进一步WebSocket 端点可以像普通 HTTP 接口一样使用Depends/Security/Cookie/Header/Path/Query配合WebSocketException与规范关闭码完成握手阶段的鉴权最后用ConnectionManager统一管理多条连接借助WebSocketDisconnect异常感知断连并向剩余客户端广播。若需继续深入了解协议层面的全部选项与更进阶的用法可进一步阅读仓库中完整的 WebSocket 教程文档法语文档 / 英文字档并对照 docs_src/websockets_ 下的三个可运行示例逐一实验。把本文示例投入生产前请务必为连接管理补充持久化订阅层与横向扩容方案并按照你的安全策略在依赖中落实鉴权校验。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastAPI WebSockets 实战指南：连接建立、消息收发、依赖注入与多客户端广播
FastAPI WebSockets 实战指南连接建立、消息收发、依赖注入与多客户端广播【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapiWebSocket 在 HTTP 之上提供一条全双工的持久化通道非常适合聊天、实时通知、协同编辑、进度推送等低延迟双向交互场景。FastAPI在其高级用法中原生支持 WebSocket你可以用装饰器直接声明 WebSocket 路由在异步端点里await收发文本、二进制与 JSON 数据还能像普通 HTTP 路径操作一样使用Depends、Security、Cookie、Header、Path、Query做参数解析与依赖注入。读完本文你将掌握在 FastAPI 中接入websockets依赖、创建并调试最小可用的 WebSocket 服务、在端点内使用依赖校验身份以及用连接管理器处理断连并向多客户端广播消息的完整方案。本文以仓库中的 WebSockets 官方文档法语版与 英文原版 内容一致为主线并结合 fastapi 源码与 docs_src/websockets_ 下的可运行教程示例展开。所有代码示例均取自教程源码可直接复制运行。WebSocket 与 FastAPI从 HTTP 请求-响应到全双工长连接普通 HTTP 接口遵循请求-响应模型客户端发起请求服务端返回响应后连接即告一段落服务端无法主动向客户端推送数据。而 WebSocket 协议在建立连接后服务端与客户端可以在同一条持久连接上随时互相发送消息。在 FastAPI 中启用 WebSocket 非常简单声明一个以app.websocket()装饰的路由路由函数接收一个WebSocket类型的参数即可在其上await接收消息、主动发送消息。FastAPI 并没有另起炉灶实现一套协议栈而是直接复用了 Starlette 的 WebSocket 实现。查看 fastapi/websockets.py 可以看到这个模块只有三行将 Starlette 的WebSocket、WebSocketDisconnect、WebSocketState原样重新导出from starlette.websockets import WebSocket as WebSocket # noqa from starlette.websockets import WebSocketDisconnect as WebSocketDisconnect # noqa from starlette.websockets import WebSocketState as WebSocketState # noqa而在 fastapi/init.py 中WebSocket、WebSocketDisconnect、WebSocketException都通过这个模块对外暴露。因此你可以直接from fastapi import WebSocket而无需关心它来自 Starlette——FastAPI 只是帮你省去了这一层导入细节。第一步安装websockets协议库WebSocket 的协议编解码需要底层库支持。在 FastAPI 中使用 WebSocket 前先把 Python 的websockets库加入项目依赖$ uv add websockets --- 100%websockets是一个以纯 Python 实现的 WebSocket 协议库负责完成握手、帧解析、消息编解码等底层工作。本文所在的仓库是一个 FastAPI 文档与代码库其依赖锁定文件 uv.lock 使用uv管理因此官方教程统一采用uv add命令添加依赖。如果你使用其他包管理器也可等价地执行pip install websockets等命令效果相同。建立一个最小 WebSocket 通道要演示 WebSocket 的服务端行为需要一个客户端。官方文档指出在真实生产环境中客户端通常是基于 React、Vue.js 或 Angular 等现代框架构建的前端直接调用其 WebSocket 工具直接与后端 WebSocket 端点通信的原生移动应用其他任何能够访问该 WebSocket 端点的客户端程序。但为了把注意力集中在服务端逻辑上教程采用了最简单的方式——把一段内嵌 JavaScript 的 HTML 放在一个超长字符串里作为演示客户端页面。这种方式不适合生产但足够让你在浏览器里立刻体验连接与收发消息的完整链路。完整示例见 docs_src/websockets_/tutorial001_py310.pyfrom fastapi import FastAPI, WebSocket from fastapi.responses import HTMLResponse app FastAPI() html ...一段内嵌 JavaScript 的 HTML 页面JavaScript 中创建 new WebSocket(ws://localhost:8000/ws) 并处理 onmessage / send app.get(/) async def get(): return HTMLResponse(html) app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text(fMessage text was: {data})其中 HTML 字符串第 6–38 行里最关键的是浏览器端逻辑var ws new WebSocket(ws://localhost:8000/ws); ws.onmessage function(event) { // 把服务端发回的消息追加到 ul idmessages 列表中 }; function sendMessage(event) { var input document.getElementById(messageText); ws.send(input.value); // 向服务端发送消息 input.value ; event.preventDefault(); }ws://localhost:8000/ws中的地址必须与你定义的 WebSocket 路由路径一致且当页面通过 HTTP 提供服务时WebSocket 也需使用对应的ws://HTTPS 下则为wss://。路由装饰器与端点签名创建 WebSocket 端点只需要两处关键代码app.websocket(/ws) # ① 声明 WebSocket 路由 async def websocket_endpoint(websocket: WebSocket): # ② 参数标注为 WebSocketapp.websocket(/ws)与app.get(/...)的使用方式一致只是走 WebSocket 协议端点函数第一个参数websocket: WebSocket就是当前连接对象。这里WebSocket由fastapi直接导出等价于from starlette.websockets import WebSocket二者是同一个类。等待与发送消息文本、二进制与 JSON进入端点函数后WebSocket 的操作是一个事件循环。教程中的核心收发逻辑为await websocket.accept() # ① 接受连接完成握手 while True: # ② 持续监听 data await websocket.receive_text() # ③ 等待客户端消息 await websocket.send_text(fMessage text was: {data}) # ④ 回发消息四个步骤各有讲究await websocket.accept()必须显式接受连接。只有在调用之后浏览器端onopen才会触发双方才能开始交换数据。while True长循环使协程持续挂起在接收操作上直到连接关闭抛异常为止。await websocket.receive_text()挂起等待并返回下一条文本消息。若客户端断开这里会抛出WebSocketDisconnect详见下文断连处理一节。await websocket.send_text(...)向客户端发送一条文本消息。除文本外同一套WebSocket对象还支持二进制数据与 JSON你可以用相应的 receive/send 系列方法收发原始字节或直接收发可被 JSON 序列化的结构化数据。也就是说文本、二进制、JSON 三类载荷都可以在这条连接上双向传输具体使用哪一组 API 取决于业务数据的形态。把教程跑起来启动服务并在浏览器中验证将完整示例代码写入main.py然后启动开发服务器$ uv run fastapi dev INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)在浏览器中打开 http://127.0.0.1:8000即可看到教程中的演示页面页面顶部是 WebSocket Chat 标题下方有一个消息输入框和 Send 按钮。在输入框中输入文字并点击发送或回车浏览器端 JavaScript 会把它经ws://localhost:8000/ws发给服务端服务端收到后回发Message text was: 你的输入ws.onmessage再把回显追加到页面的消息列表里。你可以连续发送多条消息它们全部复用同一条WebSocket 连接——这正是 WebSocket 与每次请求新建连接的 HTTP 之间的本质区别。在 WebSocket 端点中使用Depends与其他工具WebSocket 端点并不局限在收发消息上。在app.websocket端点内部你同样可以从fastapi导入并使用这些对象用法与普通路径操作完全一致Depends—— 声明并执行依赖Security—— 带安全语义的依赖Cookie—— 读取请求 CookieHeader—— 读取请求头Path—— 声明路径参数Query—— 声明查询参数教程的第二个示例 docs_src/websockets_/tutorial002_an_py310.py 演示了把认证逻辑放进依赖的做法仓库同时提供了对应的非Annotated写法 tutorial002_py310.py。先看依赖函数它同时声明了一个可选 Cookie 和一个可选查询参数async def get_cookie_or_token( websocket: WebSocket, session: Annotated[str | None, Cookie()] None, token: Annotated[str | None, Query()] None, ): if session is None and token is None: raise WebSocketException(codestatus.WS_1008_POLICY_VIOLATION) return session or token再看 WebSocket 端点它把依赖结果注入进来app.websocket(/items/{item_id}/ws) async def websocket_endpoint( *, websocket: WebSocket, item_id: str, # 来自 URL 路径 q: int | None None, # 来自查询串 cookie_or_token: Annotated[str, Depends(get_cookie_or_token)], # 依赖注入 ): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text( fSession cookie or query token value is: {cookie_or_token} ) if q is not None: await websocket.send_text(fQuery parameter q is: {q}) await websocket.send_text(fMessage text was: {data}, for item ID: {item_id})这段代码展示了三个要点路径参数照常可用item_id从/items/{item_id}/ws中解析查询参数照常可用q可直接作为函数参数声明q: int | None None也可像token那样放进依赖依赖的返回值随消息循环使用cookie_or_token在连接期间保持一致端点每收到一条消息都会引用它。注意在 WebSocket 端点中依赖的声明顺序没有强制要求websocket对象只需在参数列表中声明即可被自动注入。上述代码把websocket放进了仅限关键字的参数区*之后这只是一种风格选择。为什么这里要抛WebSocketException而不是HTTPException依赖校验失败时代码抛出的是WebSocketException(codestatus.WS_1008_POLICY_VIOLATION)。官方文档特别强调既然是 WebSocket 通道再抛HTTPException它是为 HTTP 响应设计的并没有意义正确的做法是抛出WebSocketException并使用 RFC 6455 规范 第 7.4.1 节中定义的合法关闭码——status.WS_1008_POLICY_VIOLATION1008策略违规就是其中之一。仓库源码印证了这一设计。在 fastapi/exceptions.py 中WebSocketException继承自 Starlette 的WebSocketException其构造参数是code来自规范合法关闭码表的一个整数reason可选关闭原因字符串它是 UTF-8 编码数据具体含义由应用自行解释规范并不规定其内容。同时 fastapi/init.py 将它随包导出因此示例里可以直接from fastapi import WebSocketException, status。在依赖校验场景中codestatus.WS_1008_POLICY_VIOLATION会告诉客户端本次关闭是因为策略认证不通过客户端可据此区分关闭类型。用带依赖的示例实测同样把代码放入main.py并启动uv run fastapi dev打开 http://127.0.0.1:8000页面会变成第二个示例的样子——需要先填写两个字段Item ID会拼进 URL 路径例如foo对应/items/foo/wsToken作为查询参数随连接请求发送例如some-key-token它将被依赖函数get_cookie_or_token捕获。点击 Connect 建立连接后再通过 Message 输入框发送消息。由于没有提供名为session的 Cookieget_cookie_or_token会取查询参数token作为凭据每次消息回显的第一行都会报告Session cookie or query token value is: some-key-token若同时传了查询参数q还会追加一行Query parameter q is: ...最后回显原始消息与item_id。如果把 Cookie 与 Token 都留空依赖会立即抛出WebSocketException连接将被以 1008 关闭码拒绝。处理断连与多客户端广播单条连接的收发很简单但真实应用如聊天室往往需要同时维护多条连接并向所有人广播。第三个示例 docs_src/websockets_/tutorial003_py310.py 展示了一个用内存列表管理的ConnectionManagerclass ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() # 接受连接 self.active_connections.append(websocket) # 登记连接 def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) # 移除连接 async def send_personal_message(self, message: str, websocket: WebSocket): await websocket.send_text(message) # 仅发给某个客户端 async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) # 广播给所有客户端 manager ConnectionManager()关键异常处理逻辑在端点里app.websocket(/ws/{client_id}) async def websocket_endpoint(websocket: WebSocket, client_id: int): await manager.connect(websocket) try: while True: data await websocket.receive_text() await manager.send_personal_message(fYou wrote: {data}, websocket) await manager.broadcast(fClient #{client_id} says: {data}) except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast(fClient #{client_id} left the chat)工作流程可以拆成三段登记manager.connect()接受连接并把WebSocket加入内存列表同时给该连接分配一个client_id浏览器端用Date.now()生成并显示在页面标题旁。消息循环每次收到一条文本先send_personal_message把你说了什么单独回给发送者再broadcast把某号客户端说了什么通知到列表里的所有人。异常退出某个客户端关闭页面/标签页时连接断开此时await websocket.receive_text()会抛出WebSocketDisconnect异常——它同样来自 Starlette 并通过 fastapi/websockets.py 导出可直接from fastapi import WebSocketDisconnect。except捕获后调用manager.disconnect()把它从列表移除并广播一条离开消息其余客户端会收到形如Client #1596980209979 left the chat动手试多客户端在多个浏览器标签页中打开应用从各个标签页分别发消息观察所有标签页都能收到彼此的广播然后关闭其中一个标签页——其余客户端会立刻收到某某 left the chat的广播说明断连已被正确侦测并清理。内存实现的边界何时该升级为消息总线官方文档提醒上述ConnectionManager是一个极简演示所有连接都被保存在单个进程的一块内存列表里因此它只在当前进程存活期间有效且只支持单进程部署。一旦你使用多进程如--workers 1或需要跨机器伸缩不同进程各自持有的连接列表互不可见广播就会漏发。如果你的场景需要与 FastAPI 无缝集成但又要求更强的健壮性——例如由 Redis、PostgreSQL 等外部组件做发布订阅支撑的广播方案——官方文档建议参考基于这些消息中间件的广播类库如encode/broadcaster来设计跨进程的消息分发。把连接注册表从进程内迁出、将广播语义交给 pub/sub 中间件是这类架构的通用演进方向。从源码看 FastAPI 如何承载 WebSocket 路由如果你好奇app.websocket()背后的机制可以顺着路由层追踪在 fastapi/routing.py 中定义了APIWebSocketRoute继承自 Starlette 的WebSocketRoute见第 801 行附近它负责把端点函数解析成路由 依赖 参数模型的组合当应用注册app.websocket(/ws)这类路由时约第 3047 行FastAPI 会用APIWebSocketRoute包装你的协程函数并在每个连接进入时完成参数校验与依赖求解——这正是 WebSocket 端点也能使用Depends、Cookie、Query等声明式工具的底层原因。仓库的测试也覆盖了这些能力tests/test_ws_dependencies.py 验证 WebSocket 路由上的依赖注入行为tests/test_ws_router.py 验证路由器上挂载 WebSocket 路由的行为。想确认某一种具体写法Annotated 依赖、路径参数、异常码是否可用最直接的方式就是在这些测试里找到对应的用例或照抄官方教程中的示例代码运行验证。小结本文从零到一地梳理了 FastAPI WebSocket 的完整链路先用uv add websockets引入协议库再通过app.websocket()声明路由、await websocket.accept()完成握手随后在while True循环中receive_text/send_text双向收发消息更进一步WebSocket 端点可以像普通 HTTP 接口一样使用Depends/Security/Cookie/Header/Path/Query配合WebSocketException与规范关闭码完成握手阶段的鉴权最后用ConnectionManager统一管理多条连接借助WebSocketDisconnect异常感知断连并向剩余客户端广播。若需继续深入了解协议层面的全部选项与更进阶的用法可进一步阅读仓库中完整的 WebSocket 教程文档法语文档 / 英文字档并对照 docs_src/websockets_ 下的三个可运行示例逐一实验。把本文示例投入生产前请务必为连接管理补充持久化订阅层与横向扩容方案并按照你的安全策略在依赖中落实鉴权校验。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考