一. 前言: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 走得滿可愛的。