NoneBot 钩子函数（Hook）完全指南：全局生命周期钩子与事件处理钩子实战详解

后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载本文基于 NoneBot当前仓库gh_mirrors/no/nonebot2官方文档「钩子函数」篇系统讲解 NoneBot 中两大类预定义钩子函数——全局钩子函数启动/终止/Bot 连接与断开与事件处理钩子函数事件预处理、运行预处理、API 调用钩子等的用途、注册方式与底层运行机制。读完本文你将能够熟练使用装饰器形式注册各类钩子在 NoneBot 的生命周期节点上执行数据库连接、资源清理、事件拦截、接口 Mock 等实战操作。钩子编程hooking也称作挂钩是计算机程序设计术语指通过拦截软件模块间的函数调用、消息传递、事件传递来修改或扩展操作系统、应用程序或其他软件组件的行为的各种技术。处理被拦截的函数调用、事件、消息的代码被称为钩子hook。NoneBot 正是基于这一思想预定义了一系列钩子函数供开发者在框架运行的关键节点插入自定义逻辑。这些钩子函数分为两类全局钩子函数和事件处理钩子函数全部可以通过装饰器的形式使用。钩子函数的分类与总体概览类别钩子装饰器触发时机支持依赖注入全局钩子启动准备driver.on_startupNoneBot 启动时否全局钩子终止处理driver.on_shutdownNoneBot 终止时否全局钩子Bot 连接处理driver.on_bot_connectBot 连接至 NoneBot 时是可注入Bot全局钩子Bot 断开处理driver.on_bot_disconnectBot 断开连接时是可注入Bot事件处理钩子事件预处理event_preprocessor接收到新事件时是事件处理钩子事件后处理event_postprocessor事件处理完成后是事件处理钩子运行预处理run_preprocessor运行事件响应器前是事件处理钩子运行后处理run_postprocessor运行事件响应器后是事件处理钩子平台接口调用钩子Bot.on_calling_apiBot调用平台接口时否事件处理钩子平台接口调用后钩子Bot.on_called_apiBot调用平台接口后否其中全局钩子函数针对 NoneBot 自身的运行过程由驱动器Driver负责运行因此使用前需要先获得全局驱动器即调用get_driver()。事件处理钩子函数则影响 NoneBot 的事件处理流程可以像普通的事件处理函数一样接受相应的参数通过依赖注入。全局钩子函数驱动 NoneBot 自身的生命周期全局钩子由驱动器维护并调度执行。在源码中这些钩子定义于 nonebot/internal/driver/abstract.py 的Driver基类on_startup与on_shutdown委托给内部的Lifespan管理器nonebot/internal/driver/_lifespan.py而on_bot_connect与on_bot_disconnect则将钩子函数注册进类级集合_bot_connection_hook/_bot_disconnection_hook见 abstract.py。启动准备on_startup这个钩子函数会在 NoneBot 启动时运行。很多时候我们并不希望在模块被导入时就执行一些耗时操作如连接数据库这时候我们可以在这个钩子函数中进行这些操作from nonebot import get_driver driver get_driver() driver.on_startup async def do_something(): pass从源码看Lifespan.startup()会先创建后台任务组task_group然后依次执行所有on_startup注册的函数nonebot/internal/driver/_lifespan.py。钩子函数既可以是async def异步函数也可以是普通同步函数——_run_lifespan_func内部通过is_coroutine_callable判断若是同步函数则用run_sync包装后执行见 _lifespan.py。测试用例 tests/test_driver.py 验证了多个on_startup钩子按注册顺序依次执行。终止处理on_shutdown这个钩子函数会在 NoneBot 终止时运行。我们可以在这个钩子函数中进行一些清理工作如关闭数据库连接from nonebot import get_driver driver get_driver() driver.on_shutdown async def do_something(): pass注意Lifespan.shutdown()的实现细节它会逆序执行所有on_shutdown函数以保证栈式顺序见 _lifespan.py。这意味着后注册的清理函数会先执行从而保证依赖关系的对称性——例如先断开依赖方、再关闭被依赖的资源。上述测试用例也验证了shutdown_log [2, 1]的逆序执行行为。Bot 连接处理on_bot_connect这个钩子函数会在任何协议适配器连接Bot对象至 NoneBot 时运行。支持依赖注入可以直接注入Bot对象from nonebot import get_driver driver get_driver() driver.on_bot_connect async def do_something(bot: Bot): pass在Driver._bot_connect中abstract.pyBot 对象首先被登记进self._bots字典键为bot.self_id重复连接会抛出RuntimeError随后所有连接钩子被放入任务组并发执行。源码中BOT_HOOK_PARAMS [DependParam, BotParam, DefaultParam]abstract.py限定了该钩子可注入的参数类型除了Bot对象本身还可以注入子依赖Depends和带默认值的参数。Bot 断开处理on_bot_disconnect)这个钩子函数会在Bot断开与 NoneBot 的连接时运行。支持依赖注入可以直接注入Bot对象from nonebot import get_driver driver get_driver() driver.on_bot_disconnect async def do_something(bot: Bot): passDriver._bot_disconnectabstract.py会先从_bots中移除该 Bot然后执行所有断开钩子。值得留意的是源码在此处使用了CancelScope(shieldTrue)注释确保 Bot 断开钩子总是被执行即断开钩子不会被取消操作中断——这一细节保证了清理逻辑的可靠性。连接/断开钩子均注册在Driver类级集合上因此对所有驱动实例全局生效。事件处理钩子函数介入事件处理全流程事件处理钩子定义在 nonebot/message.py 中分别存储于模块级集合_event_preprocessors、_event_postprocessors、_run_preprocessors、_run_postprocessors见 message.py。它们的注册方式为装饰器将函数解析为Dependent对象后加入对应集合message.py并且通过allow_types参数约束了各自可注入的参数类型。事件预处理event_preprocessor这个钩子函数会在 NoneBot 接收到新的事件时运行。支持依赖注入可以注入Bot对象、事件、会话状态。在这个钩子函数内抛出nonebot.exception.IgnoredException会使 NoneBot 忽略该事件from nonebot.exception import IgnoredException from nonebot.message import event_preprocessor event_preprocessor async def do_something(event: Event): if not event.is_tome(): raise IgnoredException(some reason)从事件处理主流程看handle_event在分发事件给各响应器之前会先调用_apply_event_preprocessorsmessage.py若预处理中抛出了IgnoredException则整个事件被忽略、不再分发给任何响应器message.py若抛出其他异常同样会忽略事件并记录错误日志。测试用例 tests/test_broadcast.py 专门验证了抛出IgnoredException后 matcher 不会运行。该钩子可注入的参数类型由EVENT_PCS_PARAMS限定DependParam子依赖、BotParam、EventParam、StateParam会话状态、DefaultParam带默认值的参数见 message.py 与 nonebot/typing.py。event.is_tome()是事件对象上的方法用于判断该事件是否与我Bot有关常用于群聊中过滤掉机器人自己发出的消息等场景。上述示例即实现如果事件与当前 Bot 无关则直接忽略。事件后处理event_postprocessor这个钩子函数会在 NoneBot 处理事件完成后运行。支持依赖注入可以注入Bot对象、事件、会话状态from nonebot.message import event_postprocessor event_postprocessor async def do_something(event: Event): pass在handle_event的末尾message.py所有优先级的事件响应器检查与运行完毕后会调用_apply_event_postprocessors。它可注入的参数类型与事件预处理相同EVENT_PCS_PARAMS。这里不会捕获IgnoredException仅对普通异常记录错误日志message.py。运行预处理run_preprocessor这个钩子函数会在 NoneBot 运行事件响应器前运行。支持依赖注入可以注入Bot对象、事件、事件响应器、会话状态。在这个钩子函数内抛出nonebot.exception.IgnoredException也会使 NoneBot 忽略本次运行from nonebot.message import run_preprocessor from nonebot.exception import IgnoredException run_preprocessor async def do_something(event: Event, matcher: Matcher): if not event.is_tome(): raise IgnoredException(some reason)与事件预处理不同运行预处理发生在单个事件响应器即将运行之前_run_matcher内部见 message.py因此它可以在每个 matcher 运行前分别进行校验。若抛出IgnoredException本次运行被取消但事件仍会继续分发到其他响应器。它可注入的参数类型由RUN_PREPCS_PARAMS限定message.py在事件预处理的参数基础上额外支持MatcherParam当前事件响应器与ArgParam会话参数。运行后处理run_postprocessor这个钩子函数会在 NoneBot 运行事件响应器后运行。支持依赖注入可以注入Bot对象、事件、事件响应器、会话状态、运行中产生的异常from nonebot.message import run_postprocessor run_postprocessor async def do_something(event: Event, matcher: Matcher, exception: Optional[Exception]): pass运行后处理由_apply_run_postprocessors执行message.py它接收_run_matcher捕获到的异常并作为参数传入message.py。因此你可以通过检查exception参数来判断本次运行是否失败并据此执行兜底处理。它是唯一可以注入ExceptionParam运行异常的钩子参数集合RUN_POSTPCS_PARAMS见 message.py。事件处理钩子的执行顺序总结结合 nonebot/message.py 的事件处理主流程一次事件的完整处理顺序为handle_event收到新事件执行全部事件预处理任一抛出IgnoredException则整个事件被忽略按优先级遍历所有事件响应器对每个响应器依次执行_check_matcher检查过期、权限、规则→运行预处理→ 运行响应器 →运行后处理携带异常参数全部处理完毕后执行事件后处理。平台接口调用钩子拦截与 Mock Bot API 调用平台接口调用钩子注册在Bot基类上nonebot/internal/adapter/bot.py存储于类级集合_calling_api_hook/_called_api_hook。由于是类级属性所有适配器 Bot 实例的 API 调用都会经过这些钩子。平台接口调用钩子Bot.on_calling_api这个钩子函数会在Bot对象调用平台接口时运行。在这个钩子函数中我们可以通过引起MockApiException异常来阻止Bot对象调用平台接口并返回指定的结果from nonebot.adapters import Bot from nonebot.exception import MockApiException Bot.on_calling_api async def handle_api_call(bot: Bot, api: str, data: Dict[str, Any]): if api send_msg: raise MockApiException(result{message_id: 123})其底层逻辑位于Bot.call_apibot.py在真正调用adapter._call_api之前先并发运行所有_calling_api_hook若某个钩子抛出MockApiException则捕获其result作为返回值、跳过真实 API 调用。多个钩子同时 Mock 时源码会记录警告并采用第一个钩子的结果bot.py。MockApiException定义于 nonebot/exception.py其语义正是阻止本次 API 调用或修改本次调用返回值并返回自定义内容。该钩子最典型的应用场景是测试 Mock在单元测试中拦截send_msg等接口返回伪造的message_id从而在不依赖真实平台的情况下验证业务逻辑。测试用例 tests/test_adapters/test_bot.py 中有对应的验证实现。平台接口调用后钩子Bot.on_called_api这个钩子函数会在Bot对象调用平台接口后运行。在这个钩子函数中我们可以通过引起MockApiException异常来忽略平台接口返回的结果并返回指定的结果from nonebot.adapters import Bot from nonebot.exception import MockApiException Bot.on_called_api async def handle_api_result( bot: Bot, exception: Optional[Exception], api: str, data: Dict[str, Any], result: Any, ): if not exception and api send_msg: raise MockApiException(result{**result, message_id: 123})在call_api中真实 API 调用结束后会运行所有_called_api_hookbot.py。钩子函数的五个参数依次为bot当前 Bot 对象、exception调用 API 时发生的异常无异常则为None、api调用的 API 名称、data调用的参数字典、resultAPI 的返回结果。若此处抛出MockApiException则用result覆盖原返回值并将exception重置为Nonebot.py——因此该钩子既能篡改成功调用的返回值也能在真实调用失败时兜底返回预设结果。若真实调用抛出了异常最终会在钩子执行完毕后重新抛出bot.py。与调用前钩子不同on_called_api允许在真实调用失败exception非空时通过 Mock 挽救本次调用这在实现API 降级重试失败时返回默认结果等容错逻辑时非常有用。调用钩子的调用链小结一次bot.call_api(...)的完整调用链为对应 bot.py运行全部on_calling_api钩子可 Mock 并跳过真实调用若未被跳过调用adapter._call_api执行真实平台接口异常被捕获暂存运行全部on_called_api钩子可覆盖返回值或清除异常若仍存在异常则抛出否则返回result。另外Bot基类的__getattr__将任意属性访问转换为call_api调用bot.py因此await bot.send_msg(messagehello)与await bot.call_api(send_msg, messagehello)等价两者都会经过上述钩子链。实战应用钩子函数的典型组合场景场景一在启动钩子中初始化数据库连接池将耗时操作从模块导入阶段推迟到框架启动阶段避免导入即执行、也便于在终止钩子中对称清理from nonebot import get_driver driver get_driver() _pool None driver.on_startup async def init_database(): global _pool # 在 NoneBot 启动时建立数据库连接池 _pool await create_pool(...) driver.on_shutdown async def close_database(): # 在 NoneBot 终止时关闭连接池注意 shutdown 钩子逆序执行 if _pool is not None: await _pool.close()场景二用事件预处理实现全局事件过滤例如忽略所有非 机器人的群消息from nonebot.exception import IgnoredException from nonebot.message import event_preprocessor from nonebot.adapters import Event event_preprocessor async def ignore_non_tome(event: Event): if not event.is_tome(): raise IgnoredException(not targeted at me)场景三用运行后处理实现响应器异常监控from typing import Optional from nonebot.message import run_postprocessor from nonebot.adapters import Bot, Event from nonebot.matcher import Matcher run_postprocessor async def log_matcher_error( matcher: Matcher, exception: Optional[Exception] ): if exception is not None: logger.error(fMatcher {matcher} failed: {exception!r})场景四用 API 钩子做全局限流或测试 Mock在测试环境中拦截所有发消息请求并伪造结果from nonebot.adapters import Bot from nonebot.exception import MockApiException Bot.on_calling_api async def mock_send(bot: Bot, api: str, data: dict): if api send_msg: raise MockApiException(result{message_id: 999})钩子使用注意事项全局钩子需先获取驱动器get_driver()只有在nonebot.init()被调用后才会返回已初始化的全局Driver实例nonebot/init.py否则抛出ValueError。因此在插件模块顶层直接调用get_driver()时应确保插件在nonebot.init()之后被加载。IgnoredException与MockApiException的区别前者用于事件/运行拦截事件预处理、运行预处理后者用于API 调用拦截与改写调用前/调用后钩子两者都是 nonebot/exception.py 中ProcessException的子类由 NoneBot 内部捕获处理不会向用户抛出。并发执行语义同一类型的多个钩子之间是并发执行基于anyio.create_task_group因此不要在钩子内部假设其他同类型钩子的执行顺序on_startup/on_shutdown除外它们按注册顺序/逆序依次执行。钩子抛出的普通异常事件预处理、运行预处理中的普通异常会中止对应流程并记录错误日志事件后处理、运行后处理、Bot 连接/断开钩子中的普通异常仅记录错误不影响主流程API 钩子中的普通异常会被捕获并记录真实调用照常继续调用前或结果照常返回调用后。Bot 类级注册的全局性on_calling_api/on_called_api注册在Bot类上而非实例上因此对所有协议适配器的 Bot 实例全局生效同理on_bot_connect/on_bot_disconnect注册在Driver类上对所有驱动实例全局生效。延伸阅读全局钩子依赖的驱动器体系参见 驱动器文档 与 nonebot/drivers/fastapi.py以 FastAPI 驱动为例其_lifespan_manager将 NoneBot 的Lifespan接入 FastAPI 生命周期。事件处理主流程源码nonebot/message.py。钩子相关异常定义nonebot/exception.py。相关测试用例tests/test_broadcast.py事件/运行钩子、tests/test_driver.py生命周期钩子、tests/test_adapters/test_bot.pyAPI 钩子。赞分享后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载相关推荐NoneBot2 钩子函数Hook完全指南全局生命周期钩子与事件处理钩子详解NoneBot2 钩子函数Hook完全指南全局生命周期钩子与事件处理钩子详解 本篇指南聚焦 NoneBot2 内置的钩子函数Hook体系一类由全局驱后端即时通讯NoneBot2 钩子函数Hook完全指南生命周期钩子与事件处理钩子详解NoneBot2 钩子函数Hook完全指南生命周期钩子与事件处理钩子详解 钩子编程hooking也称作“挂钩”是计算机程序设计术语指通过拦截软件后端即时通讯终极Sequelize钩子函数指南掌握Node.js ORM的生命周期事件处理终极Sequelize钩子函数指南掌握Node.js ORM的生命周期事件处理 Sequelize作为Node.js生态中最流行的ORM对象关系映射库提后端ORM数据库上一篇突破性16位Windows兼容方案在64位系统上无缝运行经典应用程序下一篇打造你的专属阅读空间Koodo Reader主题定制完全指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考