NoneBot2 websockets 客户端驱动详解：安装、连接建立与消息收发实现

后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载本篇文章基于 NoneBot2 官方 API 文档website/versioned_docs/version-2.4.4/api/drivers/websockets.md与仓库源码nonebot/drivers/websockets.py系统讲解nonebot.drivers.websockets驱动模块它的定位与适用场景、安装方式、基于Request参数的连接建立逻辑、WebSocket封装类的全部方法语义以及它在适配器调用链中的位置。读完本文你将能够独立完成该驱动的安装配置并理解如何通过 NoneBot2 的统一 WebSocket 客户端接口发起、收发与关闭连接同时掌握其与底层 websockets 库之间的参数映射关系与异常转换规则。驱动定位纯客户端 WebSocket 适配nonebot.drivers.websockets是 NoneBot2 基于 websockets 库实现的 WebSocket 客户端驱动适配模块。从源码结构看它并不实现服务端能力Mixin继承自WebSocketClientMixin见 abstract.py只提供发起客户端连接的websocket()异步上下文管理器WebSocket包装类继承自驱动层抽象的WebSocket基类见 model.py其accept()方法直接抛出NotImplementedError官方文档也明确提示本驱动仅支持客户端 WebSocket 连接:::tip提示。这意味着该驱动适合承担机器人主动向外发起 WebSocket 连接的角色例如连接第三方实时消息服务、推送网关等如果需要接收 WebSocket 连接即做服务端则应选用aiohttp、fastapi或quart等支持ASGIMixin的驱动。该模块位于 nonebot/drivers/websockets.py是 NoneBot2 驱动的典型基础驱动 能力混入组合实现。安装与启用文档给出的安装命令有两种等价方式nb driver install websockets # 或者 pip install nonebot2[websockets]nb driver install websockets是 NoneBot CLI 提供的驱动安装命令会为当前项目安装带websocketsextra 的依赖pip install nonebot2[websockets]则是直接通过 pip 安装带该可选依赖的 NoneBot2。需要特别说明的是模块源码在导入时对 websockets 库做了防御性检查见 websockets.pytry: from websockets import ClientConnection, ConnectionClosed, connect except ModuleNotFoundError as e: # pragma: no cover raise ImportError( Please install websockets first to use this driver. Install with pip: pip install nonebot2[websockets] ) from e如果未安装 websockets 库导入nonebot.drivers.websockets会直接抛出带安装提示的ImportError而不是在运行时才报错便于快速定位问题。在代码中使用该驱动时可以这样导入from nonebot.drivers.websockets import Driver as WebSocketsDriver驱动实例由 NoneBot2 框架根据配置创建一般不需要手动实例化编写自定义适配器时通过Adapter基类提供的统一websocket()接口发起连接详见下文在适配器中的调用链。连接建立Mixin.websocket 与 Request 参数映射Mixin.websocket(setup)是一个异步上下文管理器asynccontextmanager签名与返回类型如下见 websockets.pyoverride asynccontextmanager async def websocket(self, setup: Request) - AsyncGenerator[WebSocket, None]: ...参数setup是Request对象其完整字段定义见 model.py。其中与本驱动强相关的字段包括字段类型说明urlURL \| str \| RawURL目标 WebSocket 地址ws://或wss://headersHeaderTypes附加请求头以CIMultiDict存储cookiesCookieTypes附加 Cookie会被合并进请求头proxystr \| None代理地址None时按 websockets 库默认行为处理timeoutTimeoutTypes \| UnsetType超时配置支持Timeout对象、数字或未设置ping_intervalPingIntervalTypes \| UnsetType心跳 ping 间隔秒未设置时交由 websockets 库默认处理超时参数的映射规则源码将 NoneBot2 的Request.timeout翻译为 websockets 库connect()的关键字参数映射逻辑分三种情况setup.timeout是Timeout对象Timeout数据类定义见 model.py字段为total / connect / read / close / pingopen_timeout ( setup.timeout.connect or setup.timeout.read or setup.timeout.total ) timeout_kwargs { open_timeout: open_timeout, # 连接建立超时 close_timeout: setup.timeout.close, # 关闭握手超时 ping_timeout: setup.timeout.ping, # ping 响应超时 }setup.timeout是普通数字非UNSETtimeout_kwargs { open_timeout: setup.timeout, close_timeout: setup.timeout, }未设置超时UNSET回落到驱动层默认值DEFAULT_TIMEOUT定义见 model.pyDEFAULT_TIMEOUT Timeout(totalNone, connect5.0, read30.0, close10.0, ping20.0)此时open_timeout取connect(5.0) or read(30.0) or total(None)即5.0 秒close_timeout为 10.0 秒ping_timeout为 20.0 秒。注意第 1、3 种情况下open_timeout使用的是或逻辑链connect优先其次read最后total。请求头、Cookie、代理与心跳的组装在确定超时参数后驱动组装connect()的其余参数见 websockets.pykwargs exclude_unset( { **timeout_kwargs, ping_interval: setup.ping_interval, } ) connection connect( str(setup.url), additional_headers{**setup.headers, **setup.cookies.as_header(setup)}, proxysetup.proxy if setup.proxy is not None else True, **kwargs, )值得注意的细节additional_headers同时包含Request.headers与Request.cookies转换出的请求头Cookies.as_header实现见 model.py因此通过Request设置的 Cookie 会自动随连接请求发送proxy参数setup.proxy is not None时使用显式代理否则传True启用 websockets 库自身的代理发现ping_interval仅在Request显式设置时才传给connect()exclude_unset过滤掉UNSET值未设置时使用 websockets 库默认的 20 秒心跳间隔。连接建立后上下文管理器将底层ClientConnection包装为驱动层的WebSocket对象并yield给调用方退出上下文时会自动关闭连接async with connection as ws: yield WebSocket(requestsetup, websocketws)catch_closed 装饰器统一异常转换catch_closed(func)是模块级装饰器见 websockets.py用于把 websockets 库抛出的ConnectionClosed异常统一转换为 NoneBot2 自己的WebSocketClosed异常def catch_closed( func: Callable[P, CoroutineType[Any, Any, T]], ) - Callable[P, CoroutineType[Any, Any, T]]: wraps(func) async def decorator(*args: P.args, **kwargs: P.kwargs) - T: try: return await func(*args, **kwargs) except ConnectionClosed as e: raise WebSocketClosed(e.code, e.reason) return decoratorWebSocketClosed定义于 exception.py属于DriverException分支携带关闭码code与原因reason。这一层转换的意义在于适配器层只需依赖 NoneBot2 自身的异常体系无需感知底层是 websockets 还是 aiohttp 实现。装饰器作用于receive()、receive_text()、receive_bytes()三个接收方法。WebSocket 封装类方法语义详解WebSocket类包装底层 websockets 库的ClientConnection构造时接收request原Request对象与websocket连接对象并将request传给基类见 websockets.pyclass WebSocket(BaseWebSocket): override def __init__(self, *, request: Request, websocket: ClientConnection): super().__init__(requestrequest) self.websocket websocket各成员方法语义如下对照官方文档逐一说明方法签名语义与实现要点closedproperty - bool返回self.websocket.close_code is not None即底层连接是否已有关闭码用于判断连接是否已关闭accept()async - None抛出NotImplementedError。客户端连接无需接受此方法仅服务端如 aiohttp 驱动需要close(code1000, reason)async - None调用底层await self.websocket.close(code, reason)默认以 1000正常关闭关闭连接receive()async - str \| bytesawait self.websocket.recv()自动区分文本/二进制帧被catch_closed装饰receive_text()async - str接收文本帧若收到的是bytes抛TypeError(WebSocket received unexpected frame type: bytes)receive_bytes()async - bytes接收二进制帧若收到的是str抛TypeError(WebSocket received unexpected frame type: str)send_text(data)async - None发送文本帧await self.websocket.send(data)send_bytes(data)async - None发送二进制帧await self.websocket.send(data)补充说明驱动层抽象基类WebSocket见 model.py还提供了通用send(data)方法会按str/bytes自动分派到send_text/send_bytes否则抛TypeError。由于接收方法都被catch_closed装饰当连接被对端关闭后继续receive()时会得到WebSocketClosed(code, reason)异常而非底层ConnectionClosed适配器可以据此判断连接生命周期结束。Driver 组合基础驱动与能力混入模块末尾定义了Driver类见 websockets.pyif TYPE_CHECKING: class Driver(Mixin, NoneDriver): ... else: Driver combine_driver(NoneDriver, Mixin)NoneDriver是 NoneBot2 的基础驱动none类型见 none.py负责生命周期管理on_startup/on_shutdown、信号处理与任务组调度Mixin提供type websockets的驱动类型标识与websocket()客户端能力combine_driver实现见 combine.py在运行时动态生成组合类其type属性返回driver.type mixin.type形式的组合名。Driver(env, config)的构造函数参数与文档一致env为Env环境对象、config为全局Config配置对象详见 API 文档 config.md二者均由框架注入用于确定运行环境与全局配置。在适配器中的调用链统一客户端接口nonebot.drivers.websockets不是给业务插件直接使用的而是供协议适配器Adapter调用。适配器侧的统一入口定义于 adapter.pyasynccontextmanager async def websocket(self, setup: Request) - AsyncGenerator[WebSocket, None]: 建立一个 WebSocket 客户端连接请求 if not isinstance(self.driver, WebSocketClientMixin): raise TypeError(Current driver does not support websocket client) async with self.driver.websocket(setup) as ws: yield ws调用链为Adapter.websocket(setup: Request) └─ Driver.websocket(setup) # websockets 驱动 Mixin 实现 └─ websockets.connect(...) # 底层库建立连接 └─ WebSocket(request, ClientConnection) # 包装后 yield一个典型的适配器端使用模式示意from nonebot.drivers import Request async with adapter.websocket( Request(GET, wss://example.com/ws, headers{Authorization: Bearer xxx}) ) as ws: await ws.send_text(hello) data await ws.receive() await ws.close(code1000, reasondone)仓库测试 test_adapter.py 验证了这条调用链测试通过MonkeyPatch替换driver.websocket后断言adapter.websocket(request)会把同一个Request对象透传给驱动层并将驱动返回的 WebSocket 对象原样 yield。同一测试文件中websockets驱动在服务端适配器与HTTP 客户端两类测试中被标记为xfailreasonnot a server、reasonnot a http client而在test_adapter_websocket_client中正常通过——这从测试层面印证了该驱动仅具备 WebSocket 客户端能力。源码与测试参考路径读者可在当前仓库中进一步深入驱动实现nonebot/drivers/websockets.py驱动抽象基类WebSocketClientMixin、ForwardMixinnonebot/internal/driver/abstract.py数据模型Request、Timeout、DEFAULT_TIMEOUT、WebSocket基类nonebot/internal/driver/model.py驱动组合工具combine_drivernonebot/internal/driver/combine.py基础驱动NoneDrivernonebot/drivers/none.py异常定义WebSocketClosednonebot/exception.py官方 API 文档websockets.md 与驱动基类文档 index.md测试佐证test_adapter.py服务端 WebSocket 场景的测试使用 aiohttp 驱动见 test_driver.py常见问题与注意事项不能当服务端使用该驱动没有ASGIMixinWebSocket.accept()会抛NotImplementedError。需要接收对端连接的场景请改用支持服务端的驱动如aiohttp、fastapi、quart。帧类型不匹配会抛TypeError用receive_text()收到二进制帧、或用receive_bytes()收到文本帧时会抛出带明确信息的TypeError建议根据上游协议固定使用对应的接收方法。连接关闭后的行为对端关闭连接后继续receive()会得到WebSocketClosed(code, reason)异常适配器应捕获它并执行断线重连等逻辑。超时语义差异Request.timeout为数字时同时作用于open_timeout与close_timeout为Timeout对象时各超时独立生效未设置时使用驱动默认值连接 5s / 关闭 10s / ping 20s。若依赖默认值需注意长耗时握手场景可能需要显式调大。代理与心跳Request.proxy未设置时驱动传True启用 websockets 库的代理发现ping_interval未设置时沿用 websockets 库默认心跳长时间无数据传输的连接也由心跳保活。赞分享后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载相关推荐FastAPI WebSockets 实战指南连接建立、消息收发、依赖注入与多客户端广播FastAPI WebSockets 实战指南连接建立、消息收发、依赖注入与多客户端广播 WebSocket 在 HTTP 之上提供一条全双工的持久化通道非后端Web框架API设计FastAPI WebSockets 完全指南建立连接、消息收发、依赖注入与多客户端广播实战FastAPI WebSockets 完全指南建立连接、消息收发、依赖注入与多客户端广播实战 导读 本指南以 FastAPI 官方文档中的 WebSocket后端Web框架API设计NoneBot2 websockets 驱动器详解基于 websockets 库的 WebSocket 客户端驱动适配NoneBot2 websockets 驱动器详解基于 websockets 库的 WebSocket 客户端驱动适配 导读 nonebot.drivers.后端即时通讯上一篇3分钟上手Menubar应用测试与调试全攻略下一篇Roc 编译器快照测试剖析external_decl_lookup 如何验证外部模块声明的查找创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考