- 后端
- 即时通讯
- API设计
【免费下载链接】aiogram
aiogram is a modern and fully asynchronous framework for Telegram Bot API written in Python using asyncio
本文基于 aiogram 官方 API 文档 docs/api/types/update.rst 所定义的
Update对象展开,结合 aiogram/types/update.py、aiogram/dispatcher/router.py、aiogram/dispatcher/dispatcher.py 等源码实现,系统讲解 Telegram 推送的原始更新(Update)数据结构、全部可选字段、事件类型判定机制,以及 aiogram 如何把一条 Update 路由到对应的事件处理器。
Update 是什么:Telegram 与 Bot 之间的“原始事件信封”
在 Telegram Bot API 的架构里,Bot 与 Telegram 服务器之间的所有交互都围绕一个核心对象展开——Update。无论是用户发来一条普通消息、点击了内联键盘按钮、回答了投票,还是你的机器人在群里的成员状态发生变化,Telegram 服务器都会把这些事件打包成一个Update对象推送给你的机器人。aiogram 在 aiogram/types/update.py 中将其建模为一个 Pydantic 模型类,继承自TelegramObject(见 aiogram/types/base.py)。
理解Update的结构是理解 aiogram 事件处理体系的第一步,可以从以下三个关键事实入手:
- Update 是封装器,不是业务数据本身:
Update只有两个真正属于自己的核心字段——update_id和 25 个可选的“事件载荷”字段。真正的业务信息(消息文本、用户资料、回调数据等)都嵌套在载荷对象里。 - 每次更新最多只有一个事件字段非空:源码类注释明确指出 “At mostoneof the optional fields can be present in any given update”,也就是说一条 Update 要么是消息、要么是回调查询、要么是投票变更……绝不会同时携带两个业务事件。
- update_id 是防重与排序的关键:
update_id是这条更新的唯一标识符,从某个正数开始顺序递增,在使用 Webhook 时可以据此忽略重复推送的更新,或在校验顺序错乱时恢复正确的更新序列;如果超过一周没有新更新,下一个update_id会改为随机生成而不是继续顺序递增。
Update 的 26 个字段:完整清单与语义说明
依据 aiogram/types/update.py 的字段定义,Update对象包含 1 个必填字段和 25 个可选字段,下表按源码顺序完整列出:
| 字段 | 类型 | 必填/可选 | 含义 |
|---|---|---|---|
update_id | int | 必填 | 更新的唯一标识,从正数开始顺序递增;Webhook 场景用于去重与恢复顺序 |
message | Message | 可选 | 新收到的任意类型消息(文本、照片、贴纸等) |
edited_message | Message | 可选 | 机器人已知晓的消息被编辑后的新版本 |
channel_post | Message | 可选 | 新的频道帖子(任意类型) |
edited_channel_post | Message | 可选 | 频道帖子被编辑后的新版本 |
business_connection | BusinessConnection | 可选 | 机器人被连接/断开某商业账号,或用户编辑了与机器人的连接 |
business_message | Message | 可选 | 来自已连接商业账号的新消息 |
edited_business_message | Message | 可选 | 商业账号消息被编辑后的新版本 |
deleted_business_messages | BusinessMessagesDeleted | 可选 | 已连接商业账号中的消息被删除 |
guest_message | Message | 可选 | 新的访客消息,机器人可通过Message.guest_query_id与AnswerGuestQuery方法回复 |
message_reaction | MessageReactionUpdated | 可选 | 用户更改了对消息的反应;机器人必须是该聊天管理员,且必须在allowed_updates中显式指定"message_reaction"才会收到;机器人自己设置的反应不会收到此类更新 |
message_reaction_count | MessageReactionCountUpdated | 可选 | 匿名反应消息的反应数变化;同样要求机器人是管理员并在allowed_updates中显式指定"message_reaction_count";这类更新会分组推送,可能延迟数分钟 |
inline_query | InlineQuery | 可选 | 新的内联模式查询 |
chosen_inline_result | ChosenInlineResult | 可选 | 用户选择内联查询结果并发送给聊天对象(需启用反馈收集) |
callback_query | CallbackQuery | 可选 | 新的回调查询(点击内联键盘按钮) |
shipping_query | ShippingQuery | 可选 | 新的收货地址查询,仅用于灵活价格的发票 |
pre_checkout_query | PreCheckoutQuery | 可选 | 新的预结算查询,包含结算的完整信息 |
purchased_paid_media | PaidMediaPurchased | 可选 | 用户购买了机器人发送的、带有非空 payload 的付费媒体(非频道聊天) |
poll | Poll | 可选 | 新的投票状态;机器人只收到手动停止的投票和自身发送的投票更新 |
poll_answer | PollAnswer | 可选 | 用户更改了非匿名投票中的答案;机器人只收到自身发送的投票的新票数 |
my_chat_member | ChatMemberUpdated | 可选 | 机器人自身的聊天成员状态变化;私聊中仅在被封禁/解封时收到 |
chat_member | ChatMemberUpdated | 可选 | 聊天成员状态变化;机器人必须是管理员,且须在allowed_updates中显式指定"chat_member" |
chat_join_request | ChatJoinRequest | 可选 | 用户请求加入聊天;机器人须拥有can_invite_users管理员权限 |
chat_boost | ChatBoostUpdated | 可选 | 聊天获得了助推(boost);机器人须是管理员 |
removed_chat_boost | ChatBoostRemoved | 可选 | 聊天中的助推被移除;机器人须是管理员 |
managed_bot | ManagedBotUpdated | 可选 | 新创建的、由机器人管理的 Bot,或托管 Bot 的令牌/所有者发生变化 |
subscription | BotSubscriptionUpdated | 可选 | 用户付费订阅状态变化 |
stopped_message_generation | MessageGenerationStopped | 可选 | 用户请求机器人停止某条消息的生成 |
值得注意的几个容易混淆的点:
message与channel_post:前者是普通聊天中的消息,后者是频道里的帖子,二者都是Message类型但语义不同。my_chat_member与chat_member:前者关注“机器人自己”在聊天中的身份变化(如被拉入群、被封禁),后者关注“其他成员”的状态变化;chat_member需要机器人是管理员且显式订阅。message_reaction与message_reaction_count:前者是单条反应变化事件,后者是匿名反应聊天中的反应计数汇总(支持延迟推送),两者都默认不启用,需要显式加入allowed_updates。business_*系列:这些字段对应 Telegram 商业账号功能,包含连接事件、消息、编辑消息与删除消息四类。guest_message与managed_bot、subscription、stopped_message_generation:这几个是相对较新的字段,分别对应访客消息、托管 Bot、付费订阅变更与消息生成停止等能力,说明Update会随 Telegram Bot API 的演进持续扩展。
事件类型判定:event_type 与 event 属性
Update是“封装器”,那么 aiogram 如何知道该把这条更新交给哪个事件观察器?答案在Update的两个核心成员上:
event_type:类型探测属性
aiogram/types/update.py 中定义了event_type属性(带@lru_cache()缓存装饰器),它按固定优先级依次检查各个字段是否为真,返回对应的字符串标识:
@property @lru_cache() def event_type(self) -> str: if self.message: return "message" if self.edited_message: return "edited_message" if self.channel_post: return "channel_post" if self.edited_channel_post: return "edited_channel_post" # ... 依次检查 inline_query、callback_query、poll、chat_member 等全部 25 个字段 if self.stopped_message_generation: return "stopped_message_generation" raise UpdateTypeLookupError("Update does not contain any known event type.")- 返回值是字符串,例如
"message"、"callback_query"、"chat_join_request",这些字符串恰好与 Router 上注册的事件观察器名称一一对应。 - 由于前面提到“每次更新最多只有一个事件字段非空”,这种顺序检查在实际中不会产生歧义。
- 如果
Update中没有任何已知事件字段(例如 Telegram 服务器推送了一个 aiogram 尚未建模的新字段),会抛出UpdateTypeLookupError(同样定义在 aiogram/types/update.py)。
event:载荷对象属性
aiogram/types/update.py 中event属性是一个更简单的取值捷径:
@property def event(self) -> TelegramObject: return cast(TelegramObject, getattr(self, self.event_type))它借助event_type的名字,用getattr直接取出对应的载荷对象(Message、CallbackQuery、PollAnswer……),并断言为TelegramObject。在 Dispatcher 内部,event就是被分发到各个处理器的那份“业务对象”。
从 Update 到处理器:aiogram 的分发链路
Update对象定义本身只是数据层,真正的价值在于 aiogram 如何消费它。整条链路以 aiogram/dispatcher/dispatcher.py 与 aiogram/dispatcher/router.py 为核心,可以概括为“探测 → 分流 → 触发”三个阶段。
阶段一:接入(Polling / Webhook)
aiogram 支持两种更新接入方式:
- Long Polling:aiogram/dispatcher/dispatcher.py 中的
_listen_updates是“无尽更新读取器”,内部通过GetUpdates(timeout=polling_timeout, allowed_updates=allowed_updates)拉取更新,并内置了指数退避(backoff)机制,网络故障时不会让轮询进程崩溃。每收到一批更新,会把get_updates.offset = update.update_id + 1以确认已消费的更新——这正是update_id在轮询场景的典型用法(确认机制:getUpdates只返回最早未确认的 100 条更新,调用时传入大于某update_id的offset即视为确认)。 - Webhook:
Dispatcher.feed_raw_update(aiogram/dispatcher/dispatcher.py)接受原始 dict,用Update.model_validate(update, context={"bot": bot})自动完成 Dict→Update 序列化,并把处理结果作为 Webhook 响应返回;若处理器返回了一个TelegramMethod,还会触发silent_call_request自动代为执行(Webhook 机制不允许在响应中直接拿到 API 调用结果)。
阶段二:探测与分流(_listen_update)
aiogram/dispatcher/dispatcher.py 中的_listen_update是分发的枢纽:
try: update_type = update.event_type event = update.event except UpdateTypeLookupError as e: warnings.warn( "Detected unknown update type.\n" "Seems like Telegram Bot API was updated and you have " "installed not latest version of aiogram framework" f"\nUpdate: {update.model_dump_json(exclude_unset=True)}", RuntimeWarning, stacklevel=2, ) raise SkipHandler() from e kwargs.update(event_update=update) return await self.propagate_event(update_type=update_type, event=event, **kwargs)- 先通过
update.event_type得到类型字符串,再取出update.event作为业务对象,并把原始Update放入event_update上下文键——因此处理器里可以随时拿到event_update.update_id或完整的原始更新。 - 如果探测失败(未知更新类型),会发出
RuntimeWarning(提示可能 Telegram Bot API 已更新而本地 aiogram 版本过旧),并以SkipHandler()跳过本条更新,而不是让整个程序崩溃。
阶段三:事件观察器触发(propagate_event)
aiogram/dispatcher/router.py 中的propagate_event完成真正的路由:
- 以
update_type为键,从router.observers字典中取到对应的TelegramEventObserver(Router.__init__中为 25 种事件类型各预建了一个观察器,见 aiogram/dispatcher/router.py,例如self.message、self.callback_query、self.chat_join_request……)。 - 先检查观察器上的根过滤器(
check_root_filters),再调用observer.trigger(event, **kwargs)依次触发已注册的处理器,第一个所有过滤器全部通过的处理器命中后即停止传播(见 aiogram/dispatcher/event/telegram.py)。 - 若当前路由器的观察器没有处理成功,事件会继续下探到子路由器(
sub_routers),形成树状传播。Dispatcher 本身还内置了一个默认的update观察器,用于兜底处理所有未被其他观察器消费的更新。
一次事件分发生命周期
从 aiogram/dispatcher/dispatcher.py 的feed_update可以看到完整生命周期:分发前会确保Update挂载到正确的Bot实例(必要时通过Update.model_validate重建以传播 context),随后进入中间件管线,最终以日志"Update id=%s is %s. Duration %d ms by bot id=%d"记录本次更新的 id、是否被处理(handled/not handled)以及耗时(毫秒)。
allowed_updates:精准订阅你要的事件
不是所有 25 种事件都会默认推送给机器人。Telegram 的getUpdates/setWebhook都支持allowed_updates参数,aiogram 在 aiogram/client/bot.py 的get_updates(约 L1896)与set_webhook(约 L3948)中均暴露了该参数:
- 指定如
["message", "edited_channel_post", "callback_query"]时,只接收这三种更新; - 指定空列表时,接收除
chat_member、message_reaction、message_reaction_count之外的全部更新类型(默认行为); - 不指定时沿用之前的设置。
对应的更新类型枚举定义在 aiogram/enums/update_type.py 的UpdateType中,涵盖从MESSAGE到STOPPED_MESSAGE_GENERATION的全部 25 种字符串值(与event_type的返回值完全一致)。开发者可以直接引用UpdateType.MESSAGE.value这类枚举值构造allowed_updates,避免手写字符串出错。
此外,aiogram 还提供了自动解析能力:Router.resolve_used_update_types()(aiogram/dispatcher/router.py)会遍历路由链上所有观察器,收集实际注册了处理器的事件类型名,再结合内部保留的{"update", "error"}(INTERNAL_UPDATE_TYPES),生成一份“只订阅已注册事件”的allowed_updates列表,帮助减少无效流量、提升接收效率。在 examples/specify_updates.py 示例中可以看到如何在主路由与多个子路由上分别注册message、chat_member、callback_query、edited_message、my_chat_member等不同类型的事件处理器。
实战:在处理器中消费 Update 数据
理解了分发机制后,处理器的写法就非常直观了。以 examples/specify_updates.py 为骨架,演示不同Update字段对应的处理方式:
from aiogram import Bot, Dispatcher, Router from aiogram.filters import CommandStart from aiogram.types import CallbackQuery, ChatMemberUpdated, Message router = Router() # 对应 Update.message 字段:普通消息 @router.message(CommandStart()) async def command_start_handler(message: Message) -> None: await message.answer("Hello!") # 对应 Update.chat_member 字段:聊天成员状态变化 @router.chat_member() async def chat_member_update(chat_member: ChatMemberUpdated, bot: Bot) -> None: # chat_member 即 update.event;原始 Update 可通过 kwargs["event_update"] 拿到 ... # 对应 Update.callback_query 字段:内联键盘回调 @router.callback_query() async def callback_tap_me(callback_query: CallbackQuery) -> None: await callback_query.answer("OK") # 对应 Update.edited_message 字段:消息被编辑 @router.edited_message() async def edited_message_handler(edited_message: Message) -> None: await edited_message.reply("Message was edited")要点说明:
- 每个
@router.<event_type>()装饰器背后的观察器名(message、chat_member、callback_query、edited_message……)与Update.event_type的返回值一一对应,所以从 Update 字段到处理器只差一个名字。 - 处理器收到的第一个参数就是
update.event对应的载荷对象;如果还需要原始Update(例如取update_id做去重),可以通过关键字参数event_update获得,这与 tests/test_dispatcher/test_dispatcher.py 中test_listen_update测试断言的kwargs["event_update"]行为一致。 - 若更新类型未被任何观察器处理,事件会沿子路由链继续下探;最后仍无人处理时,可交由 Dispatcher 默认的
update观察器兜底(参见 docs/dispatcher/router.rst 中关于dispatcher.update()的说明)。
源码级验证:测试如何覆盖 Update 行为
aiogram 的测试套件对Update的覆盖相当完整,可作为理解行为的“活文档”:
- tests/test_dispatcher/test_dispatcher.py 的
test_listen_update使用参数化方式,为全部 25 种事件类型构造对应的Update(update_id=42, <event_type>=<载荷>),逐一验证feed_update能把载荷对象正确分发给router.observers[event_type]上注册的处理器,并断言event_update与原始Update的字段一致。 - 同文件中
test_listen_unknown_update(约 L744-L750)构造一个只有update_id、没有任何事件字段的Update(update_id=42),验证会抛出SkipHandler并伴随"Detected unknown update type"的RuntimeWarning——这正是UpdateTypeLookupError触发路径的回归测试。 _listen_updates的轮询行为也有测试覆盖(构造GetUpdates返回range(42)的更新序列并断言update.update_id逐条递增),印证了update_id + 1的确认机制。
小结
Update是 Telegram 推送事件的统一容器,必填的update_id承担去重、排序与确认职责,25 个可选字段则对应 25 种业务事件,每次推送最多命中一个。event_type/event属性完成“容器→业务对象”的解包,UpdateTypeLookupError处理未知事件并安全跳过。- Dispatcher 的
_listen_update+ Router 的propagate_event+TelegramEventObserver.trigger组成完整分发链路,处理器名即事件类型名。 - 通过
allowed_updates与resolve_used_update_types()可以精细控制接收的事件范围,提升长轮询/Webhook 效率。 - 当 Telegram Bot API 推出新能力时,
Update会新增字段,相应地event_type、UpdateType枚举与 Router 观察器也会同步扩展——保持 aiogram 版本更新即可获得新事件类型的支持。
- 后端
- 即时通讯
- API设计
【免费下载链接】aiogram
aiogram is a modern and fully asynchronous framework for Telegram Bot API written in Python using asyncio
相关推荐
aiogram StarTransaction 深入解析:Telegram Star 交易对象的字段结构、交易对手类型与获取实战
aiogram StarTransaction 深入解析:Telegram Star 交易对象的字段结构、交易对手类型与获取实战 本文以 aiogram 的 S
后端即时通讯API设计aiogram 中的 ShippingAddress 类型:字段详解与 Telegram Bot 支付配送全流程解析
aiogram 中的 ShippingAddress 类型:字段详解与 Telegram Bot 支付配送全流程解析 本文围绕 aiogram 的类型定义 Sh
后端即时通讯API设计aiogram UpdateType 枚举全解析:掌握 Telegram 全部更新类型及其在事件分发中的用法
aiogram UpdateType 枚举全解析:掌握 Telegram 全部更新类型及其在事件分发中的用法 aiogram 是构建于 Python async
后端即时通讯API设计
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考