快轉到主要內容
  1. 教學文章/

Textual Background Workers 實戰:長任務、Progress、取消與 Log Console

·9 分鐘· loading · loading · ·
Python Textual TUI Background-Workers Async Developer-Tools
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 80: 本文

featured

一. 前言:TUI 最怕的不是醜,是卡住
#

終端機工具做到一半,常常會遇到一個尷尬場景:畫面已經有按鈕、表格、狀態列,看起來很像一個認真的小工具,結果使用者按下「開始掃描」後,整個畫面凍住。

游標不動,按鈕不理你,取消鍵也沒有反應。

這時候 UI 再漂亮都沒救,因為使用者心裡只有一句話:它是不是死了?

Textual 的 background workers 就是拿來處理這種問題的。如果上一篇 Python Textual 實戰 是帶你把 TUI 做起來,這篇就是把「會跑很久的工作」安全地放進去。

今天拍拍君要做一個迷你任務面板:

  • 背景執行多步驟任務
  • 即時更新進度
  • 可以取消
  • 可以把 log 印在畫面裡
  • 錯誤不會炸掉整個 App

這篇不追求華麗 layout。我們追求一件比較工程的事情:工具在忙的時候,使用者仍然知道發生什麼事。

二. 安裝與專案目標
#

先建立一個小專案。

mkdir textual-workers-demo
cd textual-workers-demo
uv init
uv add textual

不用 uv 的話也可以用 venv:

python -m venv .venv
source .venv/bin/activate
pip install textual

這次只需要一個 app.py。我們會做一個假的資料處理任務,它不會真的打 API,也不會真的跑模型,只會模擬幾個慢步驟:掃描輸入、驗證格式、轉換資料、寫出結果、產生摘要。

為什麼用假任務?因為 background worker 的重點不是任務內容,而是 UI 跟任務之間的邊界。把這條邊界想清楚,之後換成下載檔案、跑 LLM、壓縮資料、呼叫 API,都只是替換內部函式而已。

三. 先看會卡住的版本
#

先看一個不太好的版本。

import time

from textual.app import App, ComposeResult
from textual.widgets import Button, Header, Static


class BadWorkerApp(App):
    def compose(self) -> ComposeResult:
        yield Header()
        yield Static("準備開始", id="status")
        yield Button("開始", id="start", variant="primary")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        status = self.query_one("#status", Static)
        status.update("開始處理...")
        for step in range(1, 6):
            time.sleep(1)
            status.update(f"完成第 {step}/5 步")
        status.update("完成!")

這段程式的問題很直接:on_button_pressed() 是 UI event handler,你在裡面 time.sleep(1),等於叫整個 UI 主迴圈停下來陪你睡。

結果就是畫面不會即時刷新,按鈕不能操作,快捷鍵也不一定有反應。不是 Textual 壞掉,是我們把慢工作放錯地方了。

四. 用 @work 把長任務丟到背景
#

Textual 提供 @work decorator,可以把方法變成 background worker。

import asyncio

from textual import work
from textual.app import App, ComposeResult
from textual.widgets import Button, Header, Static


class WorkerApp(App):
    def compose(self) -> ComposeResult:
        yield Header()
        yield Static("準備開始", id="status")
        yield Button("開始", id="start", variant="primary")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        if event.button.id == "start":
            self.run_job()

    @work
    async def run_job(self) -> None:
        status = self.query_one("#status", Static)
        status.update("開始處理...")
        for step in range(1, 6):
            await asyncio.sleep(1)
            status.update(f"完成第 {step}/5 步")
        status.update("完成!")

這裡有兩個關鍵差異。第一,任務方法加了 @work。第二,等待時間用 await asyncio.sleep(),不是 time.sleep()

asyncio.sleep() 會把控制權還給事件迴圈,所以 UI 可以繼續處理輸入、重繪畫面、更新狀態。這樣畫面就不會像被冷凍。

五. 建立真正的任務狀態
#

上面的版本可以跑,但我們直接在 worker 裡組 UI 字串。小範例可以,大型 App 會開始混亂。比較穩的做法是:把任務狀態整理成資料,再讓 UI 根據狀態更新。

from dataclasses import dataclass


@dataclass
class JobProgress:
    step: int
    total: int
    message: str

    @property
    def percent(self) -> int:
        return round(self.step / self.total * 100)

再準備工作步驟。

STEPS = ["掃描輸入資料", "驗證格式", "轉換資料", "寫出結果", "產生摘要"]

接著在 App 裡加一個 helper。

def update_progress(self, progress: JobProgress) -> None:
    status = self.query_one("#status", Static)
    status.update(
        f"{progress.message}\n"
        f"進度:{progress.step}/{progress.total} ({progress.percent}%)"
    )

這樣 worker 不需要到處組字串。它只要回報:現在第幾步、總共幾步、訊息是什麼。UI 怎麼呈現,是 UI 的責任。

六. 加上 ProgressBar
#

Textual 有 ProgressBar widget,比自己印百分比更直覺。

from textual.widgets import ProgressBar


def compose(self) -> ComposeResult:
    yield Header()
    yield Static("準備開始", id="status")
    yield ProgressBar(total=len(STEPS), id="progress")
    yield Button("開始", id="start", variant="primary")


def show_progress(self, progress: JobProgress) -> None:
    self.query_one("#status", Static).update(progress.message)
    self.query_one("#progress", ProgressBar).update(
        progress=progress.step,
        total=progress.total,
    )

worker 裡就可以這樣更新:

@work
async def run_job(self) -> None:
    for index, message in enumerate(STEPS, start=1):
        await asyncio.sleep(1)
        self.show_progress(JobProgress(index, len(STEPS), message))
    self.query_one("#status", Static).update("完成!")

這個版本已經比較像工具了。使用者不只知道「它還活著」,也知道「大概還剩多少」。但還有一個大洞:任務開始後,使用者不能取消。

七. 記住目前的 Worker,才能取消它
#

@work 方法被呼叫時,會回傳一個 Worker 物件。你可以把它存起來。

from textual.worker import Worker


class CancelableApp(App):
    current_worker: Worker[None] | None = None

按下開始時:

def start_job(self) -> None:
    if self.current_worker is not None and not self.current_worker.is_finished:
        self.notify("任務已經在跑了")
        return
    self.current_worker = self.run_job()

按下取消時:

def cancel_job(self) -> None:
    if self.current_worker is None or self.current_worker.is_finished:
        self.notify("沒有可以取消的任務")
        return
    self.current_worker.cancel()

取消不是魔法。你要保留 worker reference,才知道要取消誰。也要避免同時開太多份一樣的任務,不然使用者連點開始五次,你就會同時跑五個背景工作。

八. 完整版:開始、取消、進度與狀態
#

把前面的碎片組起來,核心會像這樣。

import asyncio
from dataclasses import dataclass

from textual import work
from textual.app import App, ComposeResult
from textual.widgets import Button, ProgressBar, Static
from textual.worker import Worker

STEPS = ["掃描輸入資料", "驗證格式", "轉換資料", "寫出結果", "產生摘要"]


@dataclass
class JobProgress:
    step: int
    total: int
    message: str


class JobPanelApp(App):
    current_worker: Worker[None] | None = None

    def compose(self) -> ComposeResult:
        yield Static("準備開始", id="status")
        yield ProgressBar(total=len(STEPS), id="progress")
        yield Button("開始", id="start", variant="primary")
        yield Button("取消", id="cancel", variant="error", disabled=True)

    def action_start(self) -> None:
        if self.current_worker is not None and not self.current_worker.is_finished:
            self.notify("任務已經在執行中")
            return
        self.set_running_ui(True)
        self.current_worker = self.run_job()

    def action_cancel(self) -> None:
        if self.current_worker is not None and not self.current_worker.is_finished:
            self.current_worker.cancel()

    def set_running_ui(self, running: bool) -> None:
        self.query_one("#start", Button).disabled = running
        self.query_one("#cancel", Button).disabled = not running

    def show_progress(self, progress: JobProgress) -> None:
        self.query_one("#status", Static).update(
            f"{progress.message}\n進度:{progress.step}/{progress.total}"
        )
        self.query_one("#progress", ProgressBar).update(progress=progress.step)

    @work
    async def run_job(self) -> None:
        try:
            for index, message in enumerate(STEPS, start=1):
                await asyncio.sleep(1)
                self.show_progress(JobProgress(index, len(STEPS), message))
            self.query_one("#status", Static).update("完成!")
        except asyncio.CancelledError:
            self.query_one("#status", Static).update("已取消。")
            raise
        finally:
            self.set_running_ui(False)

set_running_ui(True) 會讓開始按鈕 disabled,取消按鈕 enabled。這樣使用者不會狂開新任務。

finally 裡面會把 UI 切回可操作狀態。不管任務成功、失敗、取消,都會清理。

except asyncio.CancelledError 則是把取消狀態呈現出來。通常你應該重新 raise,讓 worker lifecycle 知道這個任務真的被取消了。不要偷偷吞掉取消,之後 debug 會很煩。

九. 加一個 Log Console
#

進度條適合回答「跑到哪裡」,Log 適合回答「剛剛發生了什麼」。Textual 有 Log widget,可以很方便地把訊息寫到畫面。

from textual.widgets import Log


def compose(self) -> ComposeResult:
    yield Static("準備開始", id="status")
    yield ProgressBar(total=len(STEPS), id="progress")
    yield Log(id="log", highlight=True)
    yield Button("開始", id="start", variant="primary")


def log(self, message: str) -> None:
    self.query_one("#log", Log).write_line(message)

任務裡就可以這樣寫:

@work
async def run_job(self) -> None:
    try:
        self.log("job started")
        for index, message in enumerate(STEPS, start=1):
            self.log(f"step {index}: {message}")
            await asyncio.sleep(1)
            self.show_progress(JobProgress(index, len(STEPS), message))
        self.log("job finished")
    except asyncio.CancelledError:
        self.log("job cancelled")
        raise
    finally:
        self.set_running_ui(False)

這樣的 UX 差很多。使用者不只看到進度,也可以看到最後卡在哪個步驟。如果你的任務會打 API,log console 也可以印重試次數、目前處理哪個檔案、哪筆資料格式不對、哪個 endpoint timeout。

十. 錯誤處理:不要讓 exception 只躲在終端機
#

背景任務最常見的壞味道之一,是錯誤只出現在開發者啟動 App 的 terminal 裡。使用者在 TUI 畫面上只看到任務停止,然後不知道發生什麼事。

拍拍君通常會把任務錯誤分成兩層:

  • 詳細錯誤寫到 log
  • 給使用者看的摘要放到 status 或 notification

範例:

@work
async def run_job(self) -> None:
    try:
        await self.process_all_steps()
    except asyncio.CancelledError:
        self.log("job cancelled by user")
        self.query_one("#status", Static).update("已取消。")
        raise
    except Exception as exc:
        self.log(f"job failed: {exc!r}")
        self.query_one("#status", Static).update("任務失敗,請查看 log。")
        self.notify("任務失敗", severity="error")
    finally:
        self.set_running_ui(False)

真實專案可以再加 traceback.format_exception()。不過不要把超長 traceback 全部塞到主要狀態列。狀態列適合簡短,Log 區域適合細節。

十一. CPU-bound 工作不要直接塞 async worker
#

如果你的慢工作是 I/O-bound,例如等 API、讀寫檔案、等資料庫、等網路下載,async worker 很適合。

但如果你的慢工作是 CPU-bound,例如壓縮大量圖片、做重型資料轉換、跑本機模型推論,你把它寫進 async worker,還是可能卡住 event loop。

這時候要考慮 @work(thread=True) 或 process。重點是:thread 裡不要直接操作 UI;如果真的要把結果送回畫面,用 call_from_thread() 把 UI 更新丟回主 thread,會比較穩。

簡單判斷方式:

  • 你在等別人回應,多半適合 async
  • 你自己在燒 CPU,多半適合 thread 或 process

這不是形式問題,是 App 會不會順的問題。

十二. 常見踩雷整理
#

第一個雷:在 async worker 裡用 time.sleep()。請改用 await asyncio.sleep()

第二個雷:沒有保存 worker reference。那你就不知道要取消哪個任務。

第三個雷:允許重複啟動同一個任務。開始前先檢查 current_worker 是否還在跑。

第四個雷:吞掉 CancelledError。如果你只是更新 UI,可以更新完後重新 raise

第五個雷:把所有訊息塞進一個 status label。狀態列、進度條、log console 分開,畫面會清楚很多。

第六個雷:CPU-bound 任務偽裝成 async。這種任務要另外處理,不然 UI 一樣會卡。

第七個雷:任務邏輯直接依賴一堆 widget。小工具可以忍,大工具會很難維護。

十三. 適合 background workers 的場景
#

Textual workers 特別適合下面這些工具:

  • 本機資料掃描器
  • API 批次檢查工具
  • CI log 觀察器
  • 檔案轉換工具
  • 小型資料清理面板
  • 本機 LLM 批次推論監看器
  • Docker 或服務健康檢查面板

共同特徵是:任務需要時間,但使用者仍然需要操作感。如果只是跑一個 0.1 秒的函式,就不用把 worker 搬出來。如果任務會超過一兩秒,就值得認真設計進度與取消。

結語:讓長任務變得可被信任
#

今天這篇的核心其實不難。

不要在 UI event handler 裡做慢工作。

@work 把任務放到背景。

Worker reference 管理生命週期。

用 ProgressBar 告訴使用者跑到哪裡。

用 cancel button 讓使用者有退路。

用 Log widget 留下可理解的過程。

Textual 的魅力不是「終端機也能漂亮」而已。它真正好用的地方,是可以把本來只會吐字串的 CLI,升級成一個有狀態、有回饋、有控制權的小型工作台。

拍拍君覺得,這也是很多內部工具剛剛好的形狀。不用開一個 Web app,也不用忍受一個完全看不出狀態的 shell script。中間這條路,Textual 走得滿可愛的。

延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 80: 本文

相關文章

Textual + SQLite 實戰:做一個終端機資料管理小工具
·8 分鐘· loading · loading
Python Textual SQLite TUI Database Developer-Tools
Python AnyIO 實戰:TaskGroup、取消管理與跨框架非同步工具
·6 分鐘· loading · loading
Python AnyIO Async Asyncio Trio Developer-Tools
Textual + DuckDB 實戰:終端機資料 Dashboard 小工具
·6 分鐘· loading · loading
Python Textual DuckDB TUI Dashboard Data-Analysis
Python Textual 實戰:終端機 TUI 應用開發完全攻略
·9 分鐘· loading · loading
Python Textual TUI Cli Terminal
Python uv build/publish 實戰:從 wheel 到 private package workflow
·11 分鐘· loading · loading
Python Uv Packaging Wheel PyPI Private-Package Developer-Tools
Streamlit + SQLModel 實戰:做一個本機 CRUD 小後台
·9 分鐘· loading · loading
Python Streamlit SQLModel SQLite CRUD Developer-Tools