Textual 事件与消息机制完全指南消息队列、冒泡、自定义消息与处理器编写【免费下载链接】textualThe lean application framework for Python. Build sophisticated user interfaces with a simple Python API. Run your apps in the terminal and a web browser.项目地址: https://gitcode.com/gh_mirrors/te/textual本文以 Textual面向 Python 的终端 UI 框架的事件系统为核心系统讲解事件Events与消息Messages的底层运行机制从消息队列与消息泵message pump的调度原理到默认行为、消息冒泡bubbling、自定义消息、临时屏蔽消息再到基于命名约定与on装饰器的处理器handler编写技巧。读完本文你将能够在自己开发的 Textual 应用中熟练地监听输入事件、向父组件上报状态变更、并写出响应迅速且结构清晰的异步事件处理代码。事件与消息先厘清概念在 Textual 中事件Events是一种特殊类型的消息Messages。事件由 Textual 框架自身在收到输入或其他状态变化时创建并发送是框架保留给内部使用的消息而开发者可以在自己的应用中创建自定义消息Custom Messages用于在组件之间协调状态、传递数据。Message基类定义在 src/textual/message.py所有事件与消息都继承自它。也就是说事件也是消息凡是对消息成立的一切对事件同样成立。因此理解消息机制就等于同时理解了事件的运作方式。接下来我们从消息从哪来、到哪里去开始。消息队列Textual 的餐厅订单模型每个 App 和 Widget 对象内部都维护着一条消息队列message queue。官方文档用餐厅订单来类比厨师接到一张订单开始做菜做菜期间陆续到达的订单排成一行厨师做完一道菜后就从队列中取下一条继续做。Textual 处理消息的方式与此完全相同消息被放入对象的消息队列由该对象内运行的asyncio 任务Task从队列中取出消息任务调用合适的处理器方法handler来处理烹饪消息。这保证了即使你的代码暂时无法立刻处理消息消息与事件也不会丢失。消息处理任务在组件挂载mount时启动持续监视队列并派发消息。对应源码位于 src/textual/message_pump.py 的_start_messages与_process_messages挂载后框架通过create_task为每个组件启动名为message pump的后台任务任务进入_process_messages_loop循环见 src/textual/message_pump.py不断从队列取消息、派发、处理直到队列关闭。一个具体的例子向 Input 中输入 Text假设你在一个Input组件中输入Text这四个字母每敲下一个键Textual 就创建一个 Key 事件并投递到该组件widget的消息队列组件后台任务从队列中取出第一条消息t的按键事件调用Input.on_key(event)方法更新界面显示这个新字母当on_key返回后Textual 继续取队列中的下一条消息重复同样的流程处理e、x、t队列清空后组件便处于空闲idle状态。需要说明的是这个例子只是为了说明机制。实际应用中事件处理速度通常足够快等你敲下一个键时前一个按键早已处理完毕因此消息队列中极少会堆积多个按键事件。值得补充的一点是在队列循环内部见_process_messages_loop源码Textual 会检查待处理消息中是否有可以合并can_replace的同类消息从而避免冗余派发队列清空时还会自动插入Idle事件让应用得以在空闲时刷新界面。默认行为处理器沿继承链自动调用如果你熟悉 Python 面向对象编程可能会想子类处理器要不要调用super()去执行基类的逻辑在 Textual 中不需要。Textual 会自动调用组件基类中定义的所有同名处理器。比如要写一个经典的 Pong 游戏Paddle组件继承自 Static当 Key 事件到达时Textual 依次调用Paddle.on_key响应方向键上/下、Static.on_key最后是Widget.on_key。这个沿 MRO 逐级查找并调用处理器的逻辑实现在 src/textual/message_pump.py 的_get_dispatch_methods中框架遍历当前类的方法解析顺序MRO把每个定义了对应处理器的类都加入调用列表且对同一个方法只派发一次。如何阻止默认行为如果你不希望继承链上的基类处理器继续执行可以在事件对象上调用 prevent_default()def on_key(self, event: events.Key) - None: event.prevent_default() # 阻止基类处理器继续执行 # ... 你自己的处理逻辑源码中prevent_default会设置消息的_no_default_action标志_get_dispatch_methods一旦检测到该标志就会立即停止遍历 MRO见if message._no_default_action: break不再调用任何基类处理器。警告prevent_default不应频繁使用。调用前务必弄清楚基类处理器做了什么否则可能禁用 Textual 内置的核心功能。消息冒泡Bubbling事件沿 DOM 向上传播消息有一个bubble类属性。若为True默认值就是True见 src/textual/message.py消息在被当前组件处理完后会被发送到其父组件继续处理。输入类事件通常都会冒泡这样当子组件不处理某个输入事件时父组件还有机会响应它。冒泡路径示例以一个包含容器和两个按钮的简化 DOM 为例当焦点在 No 按钮上时按键事件首先到达该按钮Textual 调用Button.on_key事件冒泡到按钮的父组件调用Container.on_key如果存在事件继续冒泡到其父组件App 类App 是 DOM 的根节点无处可冒传播到此结束。冒泡的底层实现在 src/textual/message_pump.py 的_on_message末尾只要消息的bubble为真、当前组件存在父节点且未调用stop()消息就会被转投_bubble_to给父组件若消息的发送者恰好就是父组件则在父组件处理一轮后自动停止冒泡避免死循环。如何停止冒泡处理器可以通过调用 stop() 来终止冒泡def on_key(self, event: events.Key) - None: event.stop() # 消息不再向上冒泡当你希望某个组件权威地处理该事件、不让它继续向上传播时就应该调用stop()。典型场景是文本输入组件当Input响应了一个按键事件后就会停止冒泡避免该按键同时触发某个键位绑定key binding。自定义消息让组件与父组件对话创建自定义消息最常见的动机是你在构建自定义组件时需要把状态变化通知给父组件。由于事件本质就是框架保留的消息自定义消息的用法与事件完全一致。完整示例可点击的颜色按钮以下示例源码见 docs/examples/events/custom01.py创建了四个颜色按钮点击后向应用发送自定义消息ColorButton.Selected应用收到后把屏幕背景色渐变为所选颜色from textual.app import App, ComposeResult from textual.color import Color from textual.message import Message from textual.widgets import Static class ColorButton(Static): A color button. class Selected(Message): Color selected message. def __init__(self, color: Color) - None: self.color color super().__init__() def __init__(self, color: Color) - None: self.color color super().__init__() def on_mount(self) - None: self.styles.margin (1, 2) self.styles.content_align (center, middle) self.styles.background Color.parse(#ffffff33) self.styles.border (tall, self.color) def on_click(self) - None: # The post_message method sends an event to be handled in the DOM self.post_message(self.Selected(self.color)) def render(self) - str: return str(self.color) class ColorApp(App): def compose(self) - ComposeResult: yield ColorButton(Color.parse(#008080)) yield ColorButton(Color.parse(#808000)) yield ColorButton(Color.parse(#E9967A)) yield ColorButton(Color.parse(#121212)) def on_color_button_selected(self, message: ColorButton.Selected) - None: self.screen.styles.animate(background, message.color, duration0.5) if __name__ __main__: app ColorApp() app.run()要点拆解自定义消息类Selected继承自 Message构造函数保存了一个 Color 对象后续处理器可以读取message.color获取所选颜色注意__init__中必须先给self.color赋值、再调用super().__init__()——若忘记调用super().__init__()post_message会抛出缺少属性错误见 src/textual/widget.py消息类定义在组件类内部。这并非强制要求但被官方推荐原因有二减少导入导入ColorButton即可通过ColorButton.Selected访问消息类建立处理器命名空间处理器名称从on_selected变为on_color_button_selected降低与其它消息的处理器重名冲突的概率。发送消息发送消息使用 post_message() 方法它把消息放入组件的消息队列并触发相应处理器。常见做法是组件向自己发送消息并允许其冒泡。这样基类也有机会处理该消息。上面的示例正是如此——ColorButton点击时self.post_message(self.Selected(self.color))发送给自己消息随后冒泡到 App这也意味着如果你继承ColorButton编写子类子类只需添加一个on_color_button_selected方法就能自己处理该消息。关于冒泡的源码细节从源码看冒泡有两个前置条件当前组件必须已挂载且父级活跃is_parent_active and is_attached并且消息的bubble标志为真。如果你自定义的消息不希望冒泡可以在子类定义时传入bubbleFalseMessage.__init_subclass__支持bubble、namespace等参数见 src/textual/message.py。临时阻止消息prevent 上下文管理器在某些场景下你可能希望暂时禁用某一类消息的投递。例如程序化更新子组件数据时不希望触发数据已变化的通知。此时可以使用 prevent() 上下文管理器配合 Python 的with关键字。示例按输入响铃、清空时不响下面的示例源码见 docs/examples/events/prevent.py在用户输入时播放终端响铃界面上的 Clear 按钮会把输入框清空——但清空操作本身也会触发Input.Changed消息如果不加处理按下 Clear 也会响铃。用prevent包裹赋值语句即可解决from textual.app import App, ComposeResult from textual.widgets import Button, Input class PreventApp(App): Demonstrates prevent context manager. def compose(self) - ComposeResult: yield Input() yield Button(Clear, idclear) def on_button_pressed(self) - None: Clear the text input. input self.query_one(Input) with input.prevent(Input.Changed): # (1) 清空输入但不发送 Input.Changed 消息 input.value def on_input_changed(self) - None: Called as the user types. self.bell() # (2) 输入时播放终端响铃 if __name__ __main__: app PreventApp() app.run()运行效果打字时终端发出响声点击 Clear 清空输入框时则安静无声。实现原理prevent内部维护一个被屏蔽消息类型的栈进入with块时把屏蔽类型入栈退出时出栈见 src/textual/message_pump.pypost_message在投递前会调用check_message_enabled检查消息是否被屏蔽见 src/textual/message_pump.py被屏蔽的消息直接丢弃。小提示现实中打字响铃非常恼人官方文档也明确不推荐在实际应用中使用此效果这里仅用于演示prevent的用法。消息处理器应用逻辑的核心Textual 应用中绝大部分逻辑都写在消息处理器里。下面深入处理器机制。处理器命名规则Textual 用如下规则把消息类映射为 Python 方法名以on_开头加上消息的命名空间如果有CamelCase 转 snake_case并以_结尾加上消息类名同样由 CamelCase 转为 snake_case。命名空间来自消息定义处的父类组件类名。例如内置的Input组件这样定义Changed消息class Input(Widget): ... class Changed(Message): Posted when the value changes. ...因为Changed是Input的子类它的命名空间是input对应处理器名为on_input_changed。这套命名空间机制让不同组件可以拥有同名消息而互不冲突。该规则在Message.__init_subclass__中实现见 src/textual/message.py框架会根据类的__qualname__推导命名空间并自动生成handler_name类变量形如on_input_changed。实用技巧如果拿不准某个消息的处理器名直接打印消息类的handler_name类变量即可 from textual.widgets import Input Input.Changed.handler_name on_input_changedon 装饰器任意命名 精确定向除了命名约定还可以用 on 装饰器把任意方法变成消息处理器。以下两种写法等价on(Button.Pressed) def handle_button_pressed(self): ... def on_button_pressed(self): ...on装饰器最大的优势是可以指定要为哪些组件处理消息。先看不使用装饰器的版本源码见 docs/examples/events/on_decorator01.py。界面上有三个按钮响铃、切换明暗主题、退出应用。单一处理器用一串if/elif按按钮 id 和 class 分发from textual.app import App, ComposeResult from textual.widgets import Button class OnDecoratorApp(App): CSS_PATH on_decorator.tcss def compose(self) - ComposeResult: Three buttons. yield Button(Bell, idbell) yield Button(Toggle dark, classestoggle dark) yield Button(Quit, idquit) def on_button_pressed(self, event: Button.Pressed) - None: Handle all button pressed events. if event.button.id bell: self.bell() elif event.button.has_class(toggle, dark): self.theme ( textual-dark if self.theme textual-light else textual-light ) elif event.button.id quit: self.exit() if __name__ __main__: app OnDecoratorApp() app.run()这种写法可用但当按钮数量增多时单一处理器里的if链会变得难以维护。on装饰器接受一个CSS 选择器选择器语法见 CSS 指南用于限定该处理器响应的组件。改造后的版本源码见 docs/examples/events/on_decorator02.py为每个按钮写独立处理器from textual import on from textual.app import App, ComposeResult from textual.widgets import Button class OnDecoratorApp(App): CSS_PATH on_decorator.tcss def compose(self) - ComposeResult: Three buttons. yield Button(Bell, idbell) yield Button(Toggle dark, classestoggle dark) yield Button(Quit, idquit) on(Button.Pressed, #bell) # (1) 匹配 id 为 bell 的按钮 def play_bell(self): Called when the bell button is pressed. self.bell() on(Button.Pressed, .toggle.dark) # (2) 同时匹配 class 为 toggle 和 dark 的按钮 def toggle_dark(self): Called when the toggle dark button is pressed. self.theme ( textual-dark if self.theme textual-light else textual-light ) on(Button.Pressed, #quit) # (3) 匹配 id 为 quit 的按钮 def quit(self): Called when the quit button is pressed. self.exit() if __name__ __main__: app OnDecoratorApp() app.run()配套样式文件见 docs/examples/events/on_decorator.tcss。两种方式中装饰器版本虽然代码行数略多但点击哪个按钮会触发什么逻辑一目了然。关于on装饰器还有几个重要细节要求control属性装饰器要求消息类提供control属性返回与该消息关联的组件。内置控件发出的消息都带有该属性如果你给自定义消息配选择器就需要在消息类中添加control属性Message.control默认返回None见 src/textual/message.py装饰器在编译期会校验control是否为内置默认实现否则抛出OnDecoratorError见 src/textual/_on.py多个装饰器命中如果多条装饰器处理器都匹配同一条消息它们会按定义顺序全部被调用而基于命名约定的处理器会在装饰器处理器之后调用此顺序实现在_get_dispatch_methods中每个类的 MRO 层级里先派发_decorated_handlers再回退到命名约定方法见 src/textual/message_pump.py若装饰器带选择器但消息缺少发送者或匹配属性不是 Widget框架会分别跳过或抛出OnNoWidget错误见 src/textual/message_pump.py。用选择器匹配消息的任意属性on装饰器还支持以关键字参数形式传入选择器去匹配消息类中列入 ALLOW_SELECTOR_MATCH 白名单的其他属性。例如只在激活的是 id 为home的标签页时才处理 TabbedContent.TabActivated 消息on(TabbedContent.TabActivated, pane#home) def home_tab(self) - None: self.log(Switched back to home tab.) ...自定义消息若想支持这种匹配需要把对应属性名加入ALLOW_SELECTOR_MATCH集合该集合默认是空集合set()否则装饰器会直接抛出OnDecoratorError见 src/textual/_on.py。处理器参数可带可不带处理器方法可以带一个位置参数也可以不带带参数时Textual 会把消息对象作为第一个参数传入。例如custom01.py中的处理器读取message.colordef on_color_button_selected(self, message: ColorButton.Selected) - None: self.screen.styles.animate(background, message.color, duration0.5)用装饰器写法同样可以接收消息on(ColorButton.Selected) def animate_background_color(self, message: ColorButton.Selected) - None: self.screen.styles.animate(background, message.color, duration0.5)如果处理器体内不需要消息中的任何信息可以省略参数。例如只想点击时响铃def on_color_button_selected(self) - None: self.app.bell()省略参数只是一种便捷写法省去一个用不到的形参。异步处理器避免阻塞 UI处理器可以是协程coroutine。给处理器加上async关键字Textual 就会await它从而允许你在处理器内await各种异步 API。如果处理器是协程多个事件理论上可以并发处理但要注意一个组件或应用必须等当前处理器返回才能从消息队列取新消息。只要处理器在几毫秒内返回UI 就几乎不会卡顿但慢速处理器会让应用变得难以使用。用餐厅类比如果来了份耗时很长的威灵顿牛排订单后续订单就会堆积、顾客只能干等——解决办法是让另一位厨师专门做牛排主厨继续接单。放到 Textual 里就是把耗时操作放到后台 asyncio 任务中执行。网络访问是慢处理器的常见来源。如果直接在处理器里同步拉取网络文件处理器可能几秒钟才返回期间组件或应用无法刷新界面。解决方案是启动一个后台 asyncio 任务来执行网络操作。示例边输入边查词典下面的示例源码见 docs/examples/events/dictionary.py 及样式 docs/examples/events/dictionary.tcss在你输入单词时通过一个词典 API 实时查询释义import asyncio try: import httpx except ImportError: raise ImportError(Please install httpx with pip install httpx ) from rich.json import JSON from textual.app import App, ComposeResult from textual.containers import VerticalScroll from textual.widgets import Input, Static class DictionaryApp(App): Searches a dictionary API as-you-type. CSS_PATH dictionary.tcss def compose(self) - ComposeResult: yield Input(placeholderSearch for a word) yield VerticalScroll(Static(idresults), idresults-container) async def on_input_changed(self, message: Input.Changed) - None: A coroutine to handle a text changed message. if message.value: # Look up the word in the background asyncio.create_task(self.lookup_word(message.value)) # 后台执行网络查询 else: # Clear the results self.query_one(#results, Static).update() async def lookup_word(self, word: str) - None: Looks up a word. url fhttps://api.dictionaryapi.dev/api/v2/entries/en/{word} async with httpx.AsyncClient() as client: results (await client.get(url)).text if word self.query_one(Input).value: self.query_one(#results, Static).update(JSON(results)) if __name__ __main__: app DictionaryApp() app.run()关键点正是高亮的那行asyncio.create_task(self.lookup_word(message.value))网络请求被放进后台协程on_input_changed立刻返回输入框保持流畅查询结果返回后再通过query_one更新结果区域并且校验了当前输入框的值仍是当初查询的单词避免过期结果覆盖新输入。运行前置条件需要先安装httpx执行pip install httpx本示例还依赖rich的JSON用于美化展示返回结果。小结Textual 的消息系统是一套队列 任务 处理器的完整事件驱动架构消息队列保证事件不丢失、按序处理src/textual/message_pump.py默认行为由框架沿 MRO 自动调用基类处理器必要时用prevent_default()截断src/textual/message.py冒泡机制让输入事件沿 DOM 向上传递用stop()权威截停src/textual/message.py自定义消息让组件与父组件解耦通信配合post_message()发送src/textual/message_pump.pyprevent上下文管理器可临时屏蔽指定类型的消息src/textual/message_pump.py命名约定 on装饰器双通道定义处理器后者支持 CSS 选择器精确定向src/textual/_on.py异步处理器 asyncio.create_task是保持 UI 响应性的关键手段。把这套机制吃透无论是编写自定义组件、组织复杂界面交互还是排查某个事件为什么没触发/触发了两次之类的问题都会事半功倍。文中所引示例均可直接在仓库 docs/examples/events 目录下运行查看效果。【免费下载链接】textualThe lean application framework for Python. Build sophisticated user interfaces with a simple Python API. Run your apps in the terminal and a web browser.项目地址: https://gitcode.com/gh_mirrors/te/textual创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考