Python Textual后台任务怎么把结果送回界面

Textual 后台任务把结果送回界面，关键不是拿到一个返回值，而是把“计算”和“更新控件”分开。对于同步库启动的线程 worker，推荐发送自定义 Message，由应用的消息处理器更新控件；如果只需要执行一次界面回调，也可以使用 call_from_thread。异步 worker 则不需要跨线程桥接，但耗时同步函数不能直接塞进 async def。

这里做一个“报表统计”小工具：点击按钮启动模拟统计，任务结束后显示记录数；再次点击代表提交新请求，旧请求即使晚到也不能覆盖最新结果。适用 Python 3.10+，Textual 需提供本文使用的 work、Message 和 Worker API。官方入口：https://textual.textualize.io/；Worker 文档：https://textual.textualize.io/guide/workers/。

把报表统计拆成任务与展示两部分

把界面想成报表的接收端，而不是后台计算的一部分。线程持有本次请求编号与统计结果，应用持有最新请求编号和 Static 控件。消息是两者的交接契约，带上 request_id、count、error 三个字段即可，后台不必知道控件的查询选择器。

本例的 ReportReady 表示一次统计结束，不承担逐条进度通知。消息处理器先比较编号，再决定显示记录数还是错误提示。这是应用层约定，不是 Textual 自动提供的“最新请求优先”保证。

建立一个可重复点击的小项目

在独立虚拟环境安装 Textual，把下列代码保存为 app.py。项目只依赖 Textual 和 Python 标准库，不访问外部接口。完成后应锁定项目实际安装的依赖版本，再集成到现有应用。

# 建立隔离环境，以下激活命令用于 macOS/Linux
python3 -m venv .venv
source .venv/bin/activate
python -m pip install textual
# Windows 请使用虚拟环境对应的激活命令

import time

from textual import work
from textual.app import App, ComposeResult
from textual.message import Message
from textual.widgets import Button, Footer, Static
from textual.worker import get_current_worker


class ReportReady(Message):
    def __init__(self, request_id: int, count: int | None, error: str | None):
        # 消息只携带普通数据，不携带控件引用
        super().__init__()
        self.request_id = request_id
        self.count = count
        self.error = error


class ReportApp(App):
    BINDINGS = [("q", "quit", "退出")]

    def __init__(self):
        super().__init__()
        self.latest_request = 0

    def compose(self) -> ComposeResult:
        yield Button("重新统计", id="reload")
        yield Static("尚未开始", id="result", markup=False)
        yield Footer()

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id != "reload":
            return
        # 编号和控件状态只在界面线程修改
        self.latest_request += 1
        request_id = self.latest_request
        self.query_one("#result", Static).update(f"任务 {request_id}：统计中")
        self.calculate_report(request_id)

    @work(thread=True, exclusive=True, group="report", exit_on_error=False)
    def calculate_report(self, request_id: int) -> None:
        worker = get_current_worker()
        try:
            # 用短暂停顿模拟同步 I/O，并提供协作取消机会
            count = 0
            for _ in range(10):
                if worker.is_cancelled:
                    return
                time.sleep(0.1)
                count += 25
        except Exception:
            # 真实项目在日志中记录异常，界面不暴露原始敏感内容
            if not worker.is_cancelled:
                self.post_message(ReportReady(request_id, None, "统计失败，请重试"))
            return
        if not worker.is_cancelled:
            self.post_message(ReportReady(request_id, count, None))

    def on_report_ready(self, message: ReportReady) -> None:
        # 已入队的旧消息也可能到达，所以接收端仍比较编号
        if message.request_id != self.latest_request:
            return
        text = message.error or f"任务 {message.request_id}：共 {message.count} 条记录"
        self.query_one("#result", Static).update(text)


if __name__ == "__main__":
    # 启动交互应用；本文没有将代码执行结果作为发布证据
    ReportApp().run()

按钮事件只准备请求并启动 worker，不等待统计结束。后台故意不调用 query_one 或 Static.update，也不修改 latest_request；线程外部数据共享越少，后续替换数据库查询或文件读取时越容易维护。

由消息处理器接收结果

post_message 是官方列出的线程安全例外，适合让后台提交数据、由界面统一展示。ReportReady 的处理器名写成 on_report_ready，和自定义消息类型对应。每次任务都创建新的消息实例，发送后不再修改其字段。

exclusive=True 用于同组 worker 的互斥替换；它不能强制停止正在执行的同步调用。本例检查 worker.is_cancelled，让旧任务在循环边界主动退出。即使线程刚检查完取消标记就遇到新请求，已经排队的消息仍可能晚到，因此 latest_request 比较不是多余代码。

处理器不把旧错误显示成新任务错误，不用“哪个线程先完成”决定当前界面。若业务要求所有任务都保留，应改成按编号存储多份记录，而不是复用这个单一结果区域。

把单次更新与任务状态分开处理

call_from_thread 必须从应用之外的线程调用，应用未运行或在应用线程内调用会抛 RuntimeError。下面是单次回调的替代片段，不要和上面的 ReportReady 发送同时使用，免得重复刷新。

# 此片段位于线程 worker 内，只传回调本身，不在后台修改控件
self.call_from_thread(self.show_report, request_id, count)

# 此方法属于 ReportApp，由桥接方法安排在应用线程执行
def show_report(self, request_id: int, count: int) -> None:
    if request_id == self.latest_request:
        self.query_one("#result", Static).update(f"共 {count} 条记录")

结果字段较多时，消息方式比散落的界面回调更便于集中维护。本例已在后台捕获统计异常并发送错误消息，因此这个错误不会再体现为 worker 的未捕获 ERROR 状态。exit_on_error=False 只是避免未处理异常导致应用退出，并不自动替你展示错误或恢复状态。

接入真实任务时控制取消与错误

把模拟循环换成阻塞 I/O 时，给底层客户端设置超时，文件资源用上下文管理器关闭。取消标记只能在代码重新获得控制权时检查，不能让一个无限阻塞的第三方调用立刻返回。CPU 密集任务还需要评估进程方案，线程 worker 不等于自动获得 CPU 并行加速。

保持消息的数据体小而稳定，不发送整份大型报表或每行一条消息。需要进度时可以按阶段或时间间隔合并通知；退出应用或离开关联界面后，停止提交新通知，也不要依赖后台结果必定可见来保存重要业务数据。

若任务使用真正的异步客户端，可以改为异步 worker 并 await 客户端方法；不要只是把 def 改成 async def 后继续调用同步阻塞库。处理当前界面也要考虑控件是否仍然存在，本例的固定主界面不涉及 Screen 切换。

运行后观察这几个交互

# 在支持交互的终端启动小项目，按 q 退出
python app.py
# 保存当前环境的实际版本，供项目依赖锁定使用
python -m pip show textual

读者本地运行时，可以先点击一次，观察统计中的提示最终变成 250 条记录；再快速连续点击，确认最终展示最新任务编号。统计期间按 q，应仍能处理退出动作。以上是示例的预期行为与观察方法，不是本文已运行得到的结果。

为什么线程 return 了，界面还是没变？

return 只是给 worker 提供结果，不会自动修改 Static。用消息处理器显式展示，或在任务状态为 SUCCESS 后读取结果。

消息发送成功等于用户已看见结果吗？

不等于。消息排队和界面绘制不是同一件事，应用退出、接收端被移除或消息被过滤时都不能当作展示成功。

后台可以直接修改 reactive 属性吗？

线程任务不应这样做。把修改安排到界面线程，消息处理器或 call_from_thread 都能表达这个边界；async worker 与线程 worker 要分别判断。

这个小项目的核心取舍是：后台只计算并报告，界面只展示当前请求。先建立清楚的消息契约，再处理取消、错误和旧结果，通常比在每个后台函数里直接找控件更容易扩展。

相关官方资料：Textual Workers 指南、App.call_from_thread API、Events and Messages 指南。