后端即时通讯API设计【免费下载链接】aiogramaiogram is a modern and fully asynchronous framework for Telegram Bot API written in Python using asyncio项目地址https://gitcode.com/gh_mirrors/ai/aiogram点击查看免费下载导读本文是 aiogram 3.x 中Router路由系统的核心技术指南。Router 是 aiogram 一切事件处理逻辑的骨架——无论是普通消息、回调查询还是复杂的多模块 Bot 工程所有 handler 都通过 Router 组织与分发。读完本文你将掌握 Router 的创建与注册方式、全部事件观察者Event Observer的用法、嵌套路由的规则与限制以及 Update 在路由树中的传播流程并能在真实项目中用 Router 搭建结构清晰的多文件 Bot。一、Router 是什么从一段最小示例开始在 aiogram 3.x 中Router负责将 Telegram 更新Update及其派生事件如 Message、CallbackQuery、Poll 等路由到对应的处理器。它既可以独立使用也可以作为Dispatcher的子路由被嵌套挂载。创建并注册一个 handler 的最小代码如下来自 docs/dispatcher/router.rstfrom aiogram import Router from aiogram.types import Message from typing import Any my_router Router(name__name__) my_router.message() async def message_handler(message: Message) - Any: await message.answer(Hello from my router!)Router(name__name__)通过__init__可传入可选的name参数用于调试时区分不同的路由实例。查看源码 aiogram/dispatcher/router.py 可以发现若未传nameRouter 会退化为hex(id(self))基于对象内存地址的十六进制字符串这也是官方示例中习惯传入name__name__的原因——保证多文件项目中每个路由都有可读的唯一标识。my_router.message()装饰器风格注册事件 handler等价于调用my_router.message.register(handler, *filters)。两种注册方式根据 Router 源码中的文档注释aiogram/dispatcher/router.py事件 handler 有两种注册途径观察者方法router.event_type.register(handler, *filters)装饰器router.event_type(*filters)两种方式完全等价。装饰器本质上就是在TelegramEventObserver.__call__中调用register方法见 aiogram/dispatcher/event/telegram.py。如果你不喜欢decorator风格可以直接调用register(...)。注意register方法不接受除flags之外的任意关键字参数。若传入多余的关键字参数会抛出UnsupportedKeywordArgument异常提示信息会明确指出“这可能是你在尝试像 2.x 版本那样注册过滤器”——这是从 aiogram 2.x 迁移到 3.x 时最常见的坑之一实现见 aiogram/dispatcher/event/telegram.py。二、事件观察者Event Observers全览Router 内部为每种 Telegram 事件类型维护一个TelegramEventObserver实例在__init__中一次性创建见 aiogram/dispatcher/router.py。每个观察者维护自己的 handler 列表、中间件和根过滤器并且在第一个匹配的 handler 执行后停止事件传播“first match wins”语义见 aiogram/dispatcher/event/telegram.py。注册 handler 的通用规则重要警告官方文档明确强调docs/dispatcher/router.rst所有 handler 必须是异步函数async defhandler 函数名无关紧要事件参数名也无严格要求但建议不要与上下文数据contextual data重名——因为 Python 函数不能接受两个同名参数。上下文数据包括bot、dispatcher、event_update、event_router等由框架注入的变量。Message 事件注意内容类型差异router.message() async def message_handler(message: types.Message) - Any: pass官方特别提醒attention块过滤 Message 事件时务必小心。不同情况下 Message 携带的属性集合完全不同——例如文本text、贴纸sticker和文档document在结构上是不同的内容类型它们分属message的不同属性。推荐的写法是先检查字段是否存在再使用最优雅的方式是使用 magic filterfrom aiogram import F router.message(F.text) # 只处理含文本的消息 async def text_handler(message: types.Message) - Any: pass router.message(F.sticker) # 只处理贴纸 async def sticker_handler(message: types.Message) - Any: passF即MagicFilter对象由 magic-filter 包提供默认随 aiogram 安装也可独立使用。F.photo的默认动作等价于lambda message: message.photo即“字段存在且不为 None”即通过还支持、!、in_、contains等操作完整用法见 docs/dispatcher/filters/magic_filters.rst。其余全部内置观察者以下所有事件均可直接以router.event_type()装饰器注册事件对象类型已标注事件观察者事件对象类型示例edited_messagetypes.Message消息被编辑channel_posttypes.Message频道新帖子edited_channel_posttypes.Message频道帖子被编辑inline_querytypes.InlineQuery内联查询chosen_inline_resulttypes.ChosenInlineResult用户选择了内联查询结果callback_querytypes.CallbackQuery回调按钮点击shipping_querytypes.ShippingQuery发货地址查询pre_checkout_querytypes.PreCheckoutQuery支付前检查polltypes.Poll投票被发送/停止poll_answertypes.PollAnswer用户回答了投票my_chat_membertypes.ChatMemberUpdatedBot 自身成员状态变化chat_membertypes.ChatMemberUpdated群成员状态变化chat_join_requesttypes.ChatJoinRequest入群申请message_reactiontypes.MessageReactionUpdated消息表情回应被更新message_reaction_counttypes.MessageReactionCountUpdated消息回应数量更新chat_boosttypes.ChatBoostUpdated频道/群组被 Boostremoved_chat_boosttypes.ChatBoostRemovedBoost 被移除errors别名errortypes.ErrorEvent其他 handler 抛出异常对应的完整注册示例来自原文档router.edited_message() async def edited_message_handler(edited_message: types.Message) - Any: pass router.channel_post() async def channel_post_handler(channel_post: types.Message) - Any: pass router.edited_channel_post() async def edited_channel_post_handler(edited_channel_post: types.Message) - Any: pass router.inline_query() async def inline_query_handler(inline_query: types.InlineQuery) - Any: pass router.chosen_inline_result() async def chosen_inline_result_handler(chosen_inline_result: types.ChosenInlineResult) - Any: pass router.callback_query() async def callback_query_handler(callback_query: types.CallbackQuery) - Any: pass router.shipping_query() async def shipping_query_handler(shipping_query: types.ShippingQuery) - Any: pass router.pre_checkout_query() async def pre_checkout_query_handler(pre_checkout_query: types.PreCheckoutQuery) - Any: pass router.poll() async def poll_handler(poll: types.Poll) - Any: pass router.poll_answer() async def poll_answer_handler(poll_answer: types.PollAnswer) - Any: pass router.my_chat_member() async def my_chat_member_handler(my_chat_member: types.ChatMemberUpdated) - Any: pass router.chat_member() async def chat_member_handler(chat_member: types.ChatMemberUpdated) - Any: pass router.chat_join_request() async def chat_join_request_handler(chat_join_request: types.ChatJoinRequest) - Any: pass router.message_reaction() async def message_reaction_handler(message_reaction: types.MessageReactionUpdated) - Any: pass router.message_reaction_count() async def message_reaction_count_handler(message_reaction_count: types.MessageReactionCountUpdated) - Any: pass router.chat_boost() async def chat_boost_handler(chat_boost: types.ChatBoostUpdated) - Any: pass router.removed_chat_boost() async def removed_chat_boost_handler(removed_chat_boost: types.ChatBoostRemoved) - Any: pass从源码看aiogram/dispatcher/router.pyaiogram 3.x 还内置了更多事件观察者deleted_business_messages、business_connection、edited_business_message、business_message、purchased_paid_media、managed_bot、guest_message、subscription、stopped_message_generation等Business/付费媒体等新 API 事件。原文档未逐一列出但这些观察者与上面表格的注册方式完全一致按需使用即可。错误事件errorserrors观察者用于捕获其他 handler 抛出的异常事件对象为types.ErrorEvent详细说明见 docs/dispatcher/errors.rstrouter.errors() async def error_handler(exception: types.ErrorEvent) - Any: pass错误处理的推荐策略是在 handler 内部用 try-except 处理可预期的异常对于全局兜底再在 router 或 dispatcher 级别注册错误处理器。路由级错误 handler 只作用于该路由内的 handlerdispatcher 级错误 handler 则作用于所有路由router.error(ExceptionTypeFilter(MyCustomException), F.update.message.as_(message)) async def handle_my_custom_exception(event: ErrorEvent, message: Message): await message.answer(Oops, something went wrong!) router.error() async def error_handler(event: ErrorEvent): logger.critical(Critical error caused by %s, event.exception, exc_infoTrue)三、嵌套路由Nested Routers模块化多文件 Bot 的基石Router 可以嵌套到另一个 Router 中从而把不同业务模块拆分到独立文件中管理。这也是大型 Bot 的标准组织方式。使用规则与两大限制官方文档以warning明确列出两条硬性限制Router 不能包含自己self-including 禁止Router 不能形成循环包含例如 router1 包含 router2、router2 包含 router3、router3 又包含 router1这是不允许的。这两条限制在源码中有完整的防御实现aiogram/dispatcher/router.pyparent_router的 setter 会依次检查——非Router实例抛ValueError已有父路由抛RuntimeError(Router is already attached to ...)self router抛RuntimeError(Self-referencing routers is not allowed)沿父链向上遍历若遇到自身抛RuntimeError(Circular referencing of Router is not allowed)。官方示例两个模块的嵌套原文档给出的多模块示例module_1.py中定义子路由module_2.py中将其挂到父路由# module_1.py router2 Router() router2.message() async def handler(...): ... # 子路由的处理器# module_2.py from module_1 import router2 router1 Router() router1.include_router(router2) # 将 router2 挂载为 router1 的子路由include_router(router)内部只做两件事类型检查后调用router.parent_router self即完成“挂载”并登记到sub_routers列表然后原样返回该 routeraiogram/dispatcher/router.py。include_routers(*routers)则是批量版本循环调用include_router且要求至少传入一个 router否则抛ValueError。实战官方多文件示例的完整组织仓库中的 examples/multi_file_bot/bot.py 展示了生产级的多文件 Bot 写法——每个 handlers 模块创建自己的 Router最后统一挂载到 Dispatcher# bot.py import asyncio import logging from os import getenv from aiogram import Bot, Dispatcher from aiogram.client.default import DefaultBotProperties from aiogram.enums import ParseMode from handlers.echo import echo_router from handlers.start import start_router TOKEN getenv(BOT_TOKEN) async def main() - None: # Dispatcher is a root router dp Dispatcher() # Register all the routers from handlers package dp.include_routers( start_router, echo_router, ) bot Bot(tokenTOKEN, defaultDefaultBotProperties(parse_modeParseMode.HTML)) await dp.start_polling(bot) if __name__ __main__: logging.basicConfig(levellogging.INFO) asyncio.run(main())各模块路由定义在 handlers/start.py 与 handlers/echo.py 中# handlers/start.py from aiogram import Router from aiogram.filters import CommandStart from aiogram.types import Message start_router Router() start_router.message(CommandStart()) async def command_start_handler(message: Message) - None: await message.answer(fHello, b{message.from_user.full_name}!/b)# handlers/echo.py from aiogram import Router from aiogram.filters import Command from aiogram.types import Message echo_router Router() echo_router.message() async def echo_handler(message: Message) - None: try: await message.send_copy(chat_idmessage.chat.id) except TypeError: await message.answer(Nice try!)这就是 Router 嵌套能力的典型价值每个业务模块命令、回声、支付、管理……各自持有 Router 与 handler彼此低耦合主程序只需一行include_routers完成装配。四、Update 事件只有根路由Dispatcher可以处理dispatcher.update() async def message_handler(update: types.Update) - Any: pass两条重要事实原文档warningnote只有根路由 Router即 Dispatcher能注册update类型的事件。原因在源码中一目了然update观察者只在Dispatcher.__init__中被创建aiogram/dispatcher/dispatcher.py普通Router实例没有这个观察者Dispatcher 自带一个默认的updatehandlerself.update.register(self._listen_update)。它的职责是把Update拆解成具体事件类型Message、CallbackQuery 等再向下传播所以你也可以注册自己的updatehandler用来兜底处理“没有被任何其他 handler 处理”的更新。Dispatcher 的parent_router属性恒为None且 setter 直接抛RuntimeError(Dispatcher can not be attached to another Router.)——Dispatcher 是根路由不能被挂到任何其他路由之下aiogram/dispatcher/dispatcher.py。Update 拆分的底层实现_listen_updateaiogram/dispatcher/dispatcher.py的工作流读取update.event_type和update.event若遇到无法识别的更新类型说明 Telegram 更新了 API 而本地 aiogram 版本较旧抛UpdateTypeLookupError记录RuntimeWarning并SkipHandler否则把event_updateupdate注入上下文数据调用self.propagate_event(update_type..., event...)完成向各观察者/子路由的传播。五、工作原理更新传播流程Update Propagation Flow路由树结构原文档给出了一个典型场景Dispatcher 下有 2 个 Router其中第 2 个 Router 又嵌套了 1 个子 Router传播过程此时一次 Update 的传播路径如下核心算法先自己、后子路由命中即停Router._propagate_eventaiogram/dispatcher/router.py是传播逻辑的核心若当前 Router 存在该update_type对应的观察者先执行根过滤器全局过滤器不通过则直接返回UNHANDLED调用观察者的trigger依次检查该观察者内的每个 handler每个 handler 自身还有过滤器若 handler 返回REJECTED哨兵按UNHANDLED处理只要拿到非UNHANDLED的响应就立即返回停止传播若本路由未处理则遍历sub_routers逐个递归调用propagate_event同样命中即 break。对应地TelegramEventObserver.triggeraiogram/dispatcher/event/telegram.py遍历自身 handler 列表handler 过滤器全部通过则执行外层包裹由 Router 链解析出的中间件若 handler 内抛SkipHandler异常则跳过继续尝试下一个 handler全部未命中返回UNHANDLED哨兵。两个关键的哨兵值定义在 aiogram/dispatcher/event/bases.pyUNHANDLED sentinel.UNHANDLED # 未处理 REJECTED sentinel.REJECTED # 被拒绝按未处理对待与中间件的协作传播过程中中间件按chain_head从子路由向根路由回溯解析并包裹 handler 调用_resolve_middlewares见 aiogram/dispatcher/event/telegram.py。仓库测试 tests/test_dispatcher/test_dispatcher.py 验证了嵌套场景下中间件的解析顺序root.outer_middleware→root.middleware→child.outer_middleware→child.middleware→ handler四个中间件各执行一次。同一文件中的test_nested_router_listen_updatetests/test_dispatcher/test_dispatcher.py则验证了最深一层子路由的 handler 能正确收到上下文包括event_router指向实际命中的router1。六、按需获取更新类型resolve_used_update_typesRouter.resolve_used_update_types(skip_eventsNone)aiogram/dispatcher/router.py用于收集整棵路由树上实际已注册 handler 的事件名称返回值是去重排序后的列表。其实现要点遍历chain_tail自身 全部子路由的深度优先序列检查每个观察者是否注册了 handler自动跳过内部事件类型{update, error}INTERNAL_UPDATE_TYPES可选skip_events参数额外排除某些事件。这一能力直接服务于长轮询的allowed_updates参数Dispatcher.start_polling在未显式传入allowed_updates时会自动调用self.resolve_used_update_types()来确定需要向 Telegram 订阅的更新类型aiogram/dispatcher/dispatcher.py。也就是说你注册了哪些 handlerBot 就只订阅对应的更新类型——既节省资源也避免收到不需要的事件。如果希望完全控制订阅范围也可以手动传入await dp.start_polling(bot, allowed_updates[message, callback_query])七、Router 的完整 API 速查Router 对外公开的核心成员aiogram/dispatcher/router.py成员说明Router(nameNone)构造函数name可选缺省用对象 id 的 hex 值include_router(router)挂载一个子路由返回该子路由会做自包含/循环引用校验include_routers(*routers)批量挂载至少传一个resolve_used_update_types(skip_eventsNone)解析当前树已注册的事件类型用于allowed_updatespropagate_event(...)/_propagate_event(...)更新传播入口与核心实现内部使用一般无需直接调用chain_head/chain_tail属性向根方向 / 向子方向的路由链生成器中间件解析与类型收集依赖它们parent_router/sub_routers父子关系父级只能通过include_router设置emit_startup/emit_shutdown递归触发各路由的启动/关闭回调Dispatcher.start_polling生命周期的一部分startup/shutdownEventObserver类型的生命周期观察者event_type观察者前述所有事件观察者见 第二部分Dispatcher继承自Router因此天然具备以上全部能力并额外提供feed_update、start_polling、run_polling、FSM 存储等根路由专属功能。八、实践建议与小结按模块拆分 Router参考 examples/multi_file_bot 的模式每个业务域一个Router(name__name__)在主入口统一include_routers用 magic filter 精确过滤 Message由于 Message 属性集合随内容类型变化优先用F.text、F.sticker等字段检查magic filter 文档而不是假设字段一定存在永远用async def并避免事件参数名与框架注入的上下文数据如bot、dispatcher冲突别让路由树成环嵌套 Router 时遵守“不包含自己、不循环包含”的红线源码会在挂载瞬间报错阻止错误处理分级业务异常尽量在 handler 内 try-except全局兜底用router.errors()/dp.errors()可结合F.update.message.as_(message)提取现场上下文让allowed_updates自动推导默认start_polling会根据注册的 handler 自动计算订阅的更新类型无需手工维护。Router 是 aiogram 3.x 事件体系的枢纽向上承接Dispatcher从 Telegram 拉取的 Update向下把每种事件精确送达对应 handler并天然支持模块化、中间件与错误兜底。掌握本文的事件观察者清单、嵌套规则与传播顺序即可构建结构清晰、易于维护的异步 Telegram Bot。相关源码与文档索引路由核心实现aiogram/dispatcher/router.py事件观察者实现aiogram/dispatcher/event/telegram.py、aiogram/dispatcher/event/bases.py根路由 Dispatcheraiogram/dispatcher/dispatcher.py多文件示例examples/multi_file_bot/bot.py、examples/multi_file_bot/handlers/start.py、examples/multi_file_bot/handlers/echo.py嵌套路由测试tests/test_dispatcher/test_dispatcher.py错误处理docs/dispatcher/errors.rst相关文档docs/dispatcher/index.rst、docs/dispatcher/dispatcher.rst、docs/dispatcher/middlewares.rst赞分享后端即时通讯API设计【免费下载链接】aiogramaiogram is a modern and fully asynchronous framework for Telegram Bot API written in Python using asyncio项目地址https://gitcode.com/gh_mirrors/ai/aiogram点击查看免费下载相关推荐aiogram Dispatcher 完全指南根路由器的更新分发、轮询与事件处理机制aiogram Dispatcher 完全指南根路由器的更新分发、轮询与事件处理机制 aiogram 的 Dispatcher 是整个事件处理体系的根路由器后端即时通讯API设计TanStack Router 路由树Route Trees完全指南文件路由、代码路由与虚拟路由的嵌套组合TanStack Router 路由树Route Trees完全指南文件路由、代码路由与虚拟路由的嵌套组合 本文是 TanStack Router 路由体前端路由SSRReact Router v8 useOutletContext 完全指南从父路由向嵌套子路由传递共享状态React Router v8 useOutletContext 完全指南从父路由向嵌套子路由传递共享状态 导读 在 React Router 的嵌套路由体系前端路由创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考