ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

NoneBot2 中的 aiohttp 驱动适配器:纯客户端 HTTP/WebSocket 连接的实现与使用

NoneBot2 中的 aiohttp 驱动适配器:纯客户端 HTTP/WebSocket 连接的实现与使用 后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载NoneBot2 的nonebot.drivers.aiohttp是基于 AIOHTTP 为主线结合 aiohttp.py 源码 与 test_driver.py 测试完整讲解它的安装方式、配置语法、Session/Mixin/WebSocket/Driver四个核心类的 API 细节、底层实现原理与实战用法。一、aiohttp 驱动在 NoneBot2 中的定位在 NoneBot2 的驱动器体系中驱动器Driver是机器人初始化的第一步负责数据的收发。驱动器按方向分为两类Forward客户端型用于 HTTP 轮询、主动连接 WebSocket 服务器与Reverse服务端型用于 WebHook、接收 WebSocket 连接参见 website/docs/advanced/driver.md。nonebot.drivers.aiohttp属于典型的客户端型Forward驱动器同时实现了 HTTPClientMixin 与 WebSocketClientMixin 两个混入能力异步发送 HTTP 请求支持自定义HTTP Method、URL、Header、Body、Cookie、Proxy、Timeout等异步建立 WebSocket 客户端连接支持自定义WebSocket URL、Header、Cookie、Proxy、Timeout等。它的派生类Driver由combine_driver(NoneDriver, Mixin)动态合并而来见 aiohttp.py 第 449 行因此继承了none驱动的信号处理、生命周期startup/shutdown hook与事件循环管理能力见 none.py 源码。从代码结构可以推断aiohttp驱动本身只负责主动发起连接不提供任何服务端监听能力这正是文档中本驱动仅支持客户端连接提示的由来。二、安装与配置1. 安装方式在 pyproject.toml 中aiohttp 驱动的可选依赖被定义为aiohttp[speedups] 3.11.0, 4.0.0。官方文档提供了两种安装方式# 方式一使用 nb-cli 安装 nb driver install aiohttp # 方式二使用 pip 安装 NoneBot2 的 aiohttp 附加依赖 pip install nonebot2[aiohttp]如果未安装 aiohttp 而直接使用该驱动aiohttp.py 第 52-58 行 会抛出带有安装提示的ImportError。2. 配置语法NoneBot 通过.env文件中的DRIVER配置项选择驱动器格式为module[:Driver][module[:Mixin]]*其中~是nonebot.drivers.的缩写见 config.py 第 410-418 行# 仅使用 aiohttp 作为唯一驱动器纯客户端 DRIVER~aiohttp # aiohttp 作为客户端混入配合 FastAPI 服务端驱动器使用 DRIVER~fastapi~aiohttp常见的配合用法是以~fastapi提供服务端 WebHook 能力再以~aiohttp或~websockets补充客户端能力。注意服务端型驱动器只能选择一个混入类驱动器只能为客户端类型aiohttp 属于客户端型因此既可以作为唯一驱动器也可以作为混入存在。3. 获取驱动器实例框架初始化完成后可通过get_driver()获取全局驱动器实例from nonebot import get_driver driver get_driver()三、核心类Session可复用的 HTTP 会话Session继承自抽象基类 HTTPClientSession封装了aiohttp.ClientSession用于复用连接、批量配置默认参数。1. 构造参数参数类型默认值说明paramsQueryTypesNone默认附加到每次请求 URL 的查询参数类型可为None、str、Mapping[str, QueryVariable]或list[tuple[str, SimpleQuery]]headersHeaderTypesNone默认请求头可为CIMultiDict[str]、dict[str, str]或list[tuple[str, str]]cookiesCookieTypesNone默认 Cookie可为None、Cookies、CookieJar、dict[str, str]或list[tuple[str, str]]versionstr \| HTTPVersionHTTPVersion.H11HTTP 协议版本HTTPVersion枚举取值H10: 1.0、H11: 1.1、H2: 2见 model.py 第 59-62 行timeoutTimeoutTypesUNSET超时配置可为float、Timeout对象或Noneproxystr \| NoneNone默认代理地址几个关键的实现细节对应 aiohttp.py 第 61-117 行构造时传入的params会被持久化为会话级默认参数在每次请求时与请求自身的params合并请求级参数优先级更高headers使用CIMultiDict大小写不敏感的多值字典保存cookies会被转换为(name, value)元组序列value为None的 Cookie 会被自动过滤version仅支持HTTPVersion.H10与HTTPVersion.H11若传入其他值如H2会抛出RuntimeError(Unsupported HTTP version: ...)——这是因为 aiohttp 底层ClientSession只接受HttpVersion10/HttpVersion11timeout若为Timeout对象则将其total、connect、read字段映射为aiohttp.ClientTimeout的同名参数若为普通数值则同时作为connect与sock_read超时。未指定时回退到全局默认值DEFAULT_TIMEOUT见 model.py 第 26 行totalNone, connect5.0, read30.0, close10.0, ping20.0。2. 实例方法async request(setup: Request) - Response发送单个 HTTP 请求返回统一封装的 Response 对象含status_code、headers、content、request。实现上会处理会话级params合并、files文件上传转换为aiohttp.FormData、Cookie 过滤、proxy与timeout回退等逻辑见 aiohttp.py 第 144-178 行。stream_request(setup: Request, *, chunk_size: int 1024) - AsyncGenerator[Response, None]发送流式请求以指定chunk_size默认 1024 字节逐块产出Response。由于 aiohttp 不保证返回固定大小的块源码中通过bytearray缓冲对response.content.iter_chunked(chunk_size)的输出进行再切分确保每个Response.content严格为chunk_size字节最后一块可能小于chunk_size空块会被跳过见 aiohttp.py 第 180-236 行相关行为在 test_aiohttp_stream_request_skip_empty_chunk 中有针对性测试。async setup() - None初始化会话创建aiohttp.ClientSessiontrust_envTrue表示信任环境变量中的代理设置并进入其异步上下文重复调用会抛出RuntimeError(Session has already been initialized)。async close() - None关闭底层aiohttp.ClientSession并将内部引用置空幂等可安全重复调用。Session支持异步上下文管理器协议async with由基类 HTTPClientSession 提供__aenter__/__aexit__进入时自动setup()退出时自动close()。四、混入类Mixin给驱动器注入客户端能力Mixin同时继承HTTPClientMixin与WebSocketClientMixin见 aiohttp.py 第 260 行即一个同时支持 HTTP 请求与 WebSocket 客户端连接的混入类。其type属性返回aiohttp在combine_driver合并后完整驱动的type会形如noneaiohttp见 combine.py 第 35-40 行。1. 方法与行为async request(setup: Request) - Response通过get_session()获取一个临时Session以async with方式发起请求并返回Response。stream_request(setup: Request, *, chunk_size1024) - AsyncGenerator[Response, None]同理以流式方式逐块产出响应。websocket(setup: Request) - AsyncGenerator[WebSocket, None]异步上下文管理器用于建立 WebSocket 客户端连接。实现要点见 aiohttp.py 第 284-348 行根据setup.version选择HttpVersion10/HttpVersion11同样只支持这两个版本timeout映射Timeout.read→ClientWSTimeout.ws_receiveTimeout.close未设置时回退total→ws_close普通数值则同时作为接收与关闭超时默认回退DEFAULT_TIMEOUTsetup.ping_interval会同时开启 aiohttp 的heartbeat与autoping若用户还额外配置了ping超时源码会输出警告aiohttp 驱动不暴露独立的 ping 超时该配置将被忽略连接成功后产出包装后的WebSocket对象退出上下文时自动关闭底层连接。get_session(paramsNone, headersNone, cookiesNone, versionHTTPVersion.H11, timeoutNone, proxyNone) - Session工厂方法返回一个全新的Session实例见 aiohttp.py 第 350-367 行。2. 与Driver的关系Driver由combine_driver(NoneDriver, Mixin)动态生成TYPE_CHECKING 分支下声明为class Driver(Mixin, NoneDriver)见 aiohttp.py 第 444-450 行。因此driver.request(...)、driver.stream_request(...)、driver.websocket(...)均直接可用其实现即委托给上面的Mixin方法同时继承了NoneDriver的run()、on_startup()、on_shutdown()、on_bot_connect()等生命周期管理。在 test_driver.py 中aiohttp 驱动与 websockets 驱动被并列为客户端驱动的参数化测试对象如pytest.param(nonebot.drivers.aiohttp:Driver, idaiohttp)覆盖了stream_request分块、WebSocket 连接、超时、ping 间隔等场景见 tests/test_driver.py#L326-L374 与 tests/test_driver.py#L647-L656。五、包装类WebSocketaiohttp 客户端 WebSocket 的封装WebSocket继承自抽象基类 WebSocket是aiohttp.ClientWebSocketResponse的包装器构造参数为requestRequest本次 WebSocket 连接请求sessionaiohttp.ClientSession承载连接的会话websocketaiohttp.ClientWebSocketResponseaiohttp 底层连接对象。实例方法方法返回说明async accept()untyped未实现客户端连接无需接受握手调用会抛出NotImplementedError这是客户端型 WebSocket 与 服务端 WebSocket 的关键差异async close(code1000, reason)untyped以指定关闭码与原因关闭底层连接随后关闭会话reason 按 UTF-8 编码发送async receive()str接收一帧TEXT/BINARY类型均可返回原始数据async receive_text()str仅接收文本帧收到非TEXT帧抛出TypeErrorasync receive_bytes()bytes仅接收二进制帧收到非BINARY帧抛出TypeErrorasync send_text(data: str)None发送文本帧async send_bytes(data: bytes)None发送二进制帧实现细节见 aiohttp.py 第 370-441 行closed属性直接透传底层websocket.closed私有方法_receive()统一处理接收当收到CLOSE/CLOSING/CLOSED帧时抛出nonebot.exception.WebSocketClosed并携带底层关闭码默认 1006test_aiohttp_websocket_close_frame 测试了这三种关闭帧场景close()同时关闭会话避免连接泄漏基类还提供了便捷方法send(data: str | bytes)根据数据类型自动分发到send_text/send_bytes见 model.py 第 205-212 行。六、实战示例在插件中使用 aiohttp 驱动1. 发起 HTTP 请求from nonebot import get_driver from nonebot.drivers import Request driver get_driver() async def fetch_example() - None: req Request( GET, https://example.com/api, params{page: 1}, # 查询参数 headers{Accept: application/json}, timeout10, # 10 秒超时 ) resp await driver.request(req) print(resp.status_code, resp.content.decode(utf-8))2. 流式下载分块处理async def stream_download() - None: req Request(GET, https://example.com/large.bin) async for chunk in driver.stream_request(req, chunk_size4096): # 每个 chunk.content 严格为 4096 字节末块除外 await process_chunk(chunk.content)3. 建立 WebSocket 客户端连接from nonebot.drivers import Request async def ws_client() - None: req Request( GET, wss://example.com/ws, headers{Authorization: Bearer token}, timeout30, ping_interval20, # 开启心跳每 20 秒自动 ping ) async with driver.websocket(req) as ws: await ws.send_text(hello) data await ws.receive_text() print(data)4. 复用Session批量请求async with driver.get_session( headers{User-Agent: nonebot/2.x}, timeout15, ) as session: resp1 await session.request(Request(GET, https://example.com/a)) resp2 await session.request(Request(GET, https://example.com/b))七、注意事项与限制仅客户端aiohttp 驱动不支持任何服务端功能WebHook、服务端 WebSocket 接收等需要服务端能力时请配合~fastapi/~quart等 ASGI 驱动器使用。HTTP 版本受限Session与Mixin.websocket均只支持 HTTP/1.0 与 HTTP/1.1传入HTTPVersion.H2会抛出RuntimeError。ping 超时不可用aiohttp 不暴露独立的 ping 超时配置设置Timeout.ping只会触发警告并被忽略。客户端accept()未实现客户端 WebSocket 包装器的accept()会抛出NotImplementedError属于预期行为。依赖版本本驱动依赖aiohttp[speedups] 3.11.0, 4.0.0安装时请确保版本位于该区间。八、相关资源API 文档website/versioned_docs/version-2.4.3/api/drivers/aiohttp.md、驱动基类文档 website/versioned_docs/version-2.4.3/api/drivers/index.md源码实现nonebot/drivers/aiohttp.py抽象基类与数据模型nonebot/internal/driver/abstract.py、nonebot/internal/driver/model.py驱动合并机制nonebot/internal/driver/combine.py驱动选择指南website/docs/advanced/driver.md测试用例tests/test_driver.py赞分享后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载相关推荐Emacs-wgrep性能优化处理大型项目搜索结果的7个实用技巧Emacs wgrep性能优化处理大型项目搜索结果的7个实用技巧 Emacs wgrep是Emacs编辑器中一款强大的可写grep缓冲区工具它允许您直接在搜后端即时通讯从理论到实践pull-stream设计目标与实现原理的终极指南从理论到实践pull stream设计目标与实现原理的终极指南 在JavaScript的流处理领域pull stream以其独特的设计理念和极简的实现方式脱后端Serial Studio 的 WebSocket 与 HTTP 客户端实现Network 驱动四路拆分设计解析Serial Studio 的 WebSocket 与 HTTP 客户端实现Network 驱动四路拆分设计解析 本文围绕 Serial Studio 的 S桌面应用数据可视化物联网上一篇redux-observable 与 UI 框架无关如何将业务逻辑与视图层解耦下一篇npkill版本升级指南从0.x到1.x的迁移步骤创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表