textual-tui

Textual TUI 开发

使用 Textual（一个用于构建交互式 TUI 应用的现代 Python 框架），打造可用于生产环境的终端用户界面。

快速入门

安装 Textual：

pip install textual textual-dev

基本应用结构：

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

class MyApp(App):
    """A simple Textual app."""
    
    def compose(self) -> ComposeResult:
        """Create child widgets."""
        yield Header()
        yield Button("Click me!", id="click")
        yield Footer()
    
    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Handle button press."""
        self.exit()

if __name__ == "__main__":
    app = MyApp()
    app.run()

开发期间启用热重载：

textual run --dev your_app.py

使用 Textual 控制台进行调试：

textual console

核心架构

应用生命周期

1. 初始化：使用配置创建 App 实例
2. 组合：通过 compose() 方法构建组件树
3. 挂载：将组件挂载到 DOM
4. 运行：事件循环处理消息并渲染 UI
5. 关闭：清理并退出

消息传递系统

Textual 使用异步消息队列处理所有交互：

from textual.message import Message

class CustomMessage(Message):
    """Custom message with data."""
    def __init__(self, value: int) -> None:
        self.value = value
        super().__init__()

class MyWidget(Widget):
    def on_click(self) -> None:
        # Post message to parent
        self.post_message(CustomMessage(42))

class MyApp(App):
    def on_custom_message(self, message: CustomMessage) -> None:
        # Handle message with naming convention: on_{message_name}
        self.log(f"Received: {message.value}")

响应式编程

使用响应式属性自动更新 UI：

from textual.reactive import reactive

class Counter(Widget):
    count = reactive(0)  # Reactive attribute
    
    def watch_count(self, new_value: int) -> None:
        """Called automatically when count changes."""
        self.refresh()
    
    def increment(self) -> None:
        self.count += 1  # Triggers watch_count

布局系统

容器布局

Textual 提供灵活的布局选项：

垂直布局（默认）：

def compose(self) -> ComposeResult:
    yield Label("Top")
    yield Label("Bottom")

水平布局：

class MyApp(App):
    CSS = """
    Screen {
        layout: horizontal;
    }
    """

网格布局：

class MyApp(App):
    CSS = """
    Screen {
        layout: grid;
        grid-size: 3 2;  /* 3 columns, 2 rows */
    }
    """

尺寸与定位

控制组件尺寸：

class MyApp(App):
    CSS = """
    #sidebar {
        width: 30;      /* Fixed width */
        height: 100%;   /* Full height */
    }
    
    #content {
        width: 1fr;     /* Remaining space */
    }
    
    .compact {
        height: auto;   /* Size to content */
    }
    """

使用 CSS 设置样式

Textual 使用类似 CSS 的语法设置样式。

内联样式

class StyledWidget(Widget):
    DEFAULT_CSS = """
    StyledWidget {
        background: $primary;
        color: $text;
        border: solid $accent;
        padding: 1 2;
        margin: 1;
    }
    """

外部 CSS 文件

class MyApp(App):
    CSS_PATH = "app.tcss"  # Load from file

颜色系统

使用 Textual 的语义颜色：

.error { background: $error; }
.success { background: $success; }
.warning { background: $warning; }
.primary { background: $primary; }

或者定义自定义颜色：

.custom {
    background: #1e3a8a;
    color: rgb(255, 255, 255);
}

常用组件

输入与表单

from textual.widgets import Input, Button, Select
from textual.containers import Container

def compose(self) -> ComposeResult:
    with Container(id="form"):
        yield Input(placeholder="Enter name", id="name")
        yield Select(options=[("A", 1), ("B", 2)], id="choice")
        yield Button("Submit", variant="primary")

def on_button_pressed(self, event: Button.Pressed) -> None:
    name = self.query_one("#name", Input).value
    choice = self.query_one("#choice", Select).value

数据显示

from textual.widgets import DataTable, Tree, Log

# DataTable for tabular data
table = DataTable()
table.add_columns("Name", "Age", "City")
table.add_row("Alice", 30, "NYC")

# Tree for hierarchical data
tree = Tree("Root")
tree.root.add("Child 1")
tree.root.add("Child 2")

# Log for streaming output
log = Log(auto_scroll=True)
log.write_line("Log entry")

容器与布局

from textual.containers import (
    Container, Horizontal, Vertical,
    Grid, ScrollableContainer
)

def compose(self) -> ComposeResult:
    with Vertical():
        yield Header()
        with Horizontal():
            with Container(id="sidebar"):
                yield Label("Menu")
            with ScrollableContainer(id="content"):
                yield Label("Content...")
        yield Footer()

事件处理

内置事件

from textual.events import Key, Click, Mount

def on_mount(self) -> None:
    """Called when widget is mounted."""
    self.log("Widget mounted!")

def on_key(self, event: Key) -> None:
    """Handle all key presses."""
    if event.key == "q":
        self.app.exit()

def on_click(self, event: Click) -> None:
    """Handle mouse clicks."""
    self.log(f"Clicked at {event.x}, {event.y}")

组件专用处理程序

def on_input_submitted(self, event: Input.Submitted) -> None:
    """Handle input submission."""
    self.query_one(Log).write(event.value)

def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
    """Handle table row selection."""
    row_key = event.row_key

键盘绑定

class MyApp(App):
    BINDINGS = [
        ("q", "quit", "Quit"),
        ("d", "toggle_dark", "Toggle dark mode"),
        ("ctrl+s", "save", "Save"),
    ]
    
    def action_quit(self) -> None:
        self.exit()
    
    def action_toggle_dark(self) -> None:
        self.dark = not self.dark

高级模式

自定义组件

创建可复用的组件：

from textual.widget import Widget
from textual.widgets import Label, Button

class StatusCard(Widget):
    """A card showing status info."""
    
    def __init__(self, title: str, status: str) -> None:
        super().__init__()
        self.title = title
        self.status = status
    
    def compose(self) -> ComposeResult:
        yield Label(self.title, classes="title")
        yield Label(self.status, classes="status")

Worker 与后台任务

重要：对于任何长时间运行的操作，都要使用 Worker，以避免阻塞 UI。事件循环必须保持响应。

Worker 基本用法

在后台线程中运行任务：

from textual.worker import Worker, WorkerState

class MyApp(App):
    def on_button_pressed(self, event: Button.Pressed) -> None:
        # Start background task
        self.run_worker(self.process_data(), exclusive=True)
    
    async def process_data(self) -> str:
        """Long-running task."""
        # Simulate work
        await asyncio.sleep(5)
        return "Processing complete"

带进度更新的 Worker

处理过程中更新 UI：

from textual.widgets import ProgressBar

class MyApp(App):
    def compose(self) -> ComposeResult:
        yield ProgressBar(total=100, id="progress")
    
    def on_mount(self) -> None:
        self.run_worker(self.long_task())
    
    async def long_task(self) -> None:
        """Task with progress updates."""
        progress = self.query_one(ProgressBar)
        
        for i in range(100):
            await asyncio.sleep(0.1)
            progress.update(progress=i + 1)
            # Use call_from_thread for thread safety
            self.call_from_thread(progress.update, progress=i + 1)

Worker 通信模式

使用 call_from_thread 以线程安全的方式更新 UI：

import time
from threading import Thread

class MyApp(App):
    def on_mount(self) -> None:
        self.run_worker(self.fetch_data(), thread=True)
    
    def fetch_data(self) -> None:
        """CPU-bound task in thread."""
        # Blocking operation
        result = expensive_computation()
        
        # Update UI safely from thread
        self.call_from_thread(self.display_result, result)
    
    def display_result(self, result: str) -> None:
        """Called on main thread."""
        self.query_one("#output").update(result)

取消 Worker

不再需要时取消 Worker：

class MyApp(App):
    worker: Worker | None = None
    
    def start_task(self) -> None:
        # Store worker reference
        self.worker = self.run_worker(self.long_task())
    
    def cancel_task(self) -> None:
        # Cancel running worker
        if self.worker and not self.worker.is_finished:
            self.worker.cancel()
            self.notify("Task cancelled")
    
    async def long_task(self) -> None:
        for i in range(1000):
            await asyncio.sleep(0.1)
            # Check if cancelled
            if self.worker.is_cancelled:
                return

Worker 错误处理

妥善处理 Worker 故障：

class MyApp(App):
    def on_mount(self) -> None:
        worker = self.run_worker(self.risky_task())
        worker.name = "data_processor"  # Name for debugging
    
    async def risky_task(self) -> str:
        """Task that might fail."""
        try:
            result = await fetch_from_api()
            return result
        except Exception as e:
            self.notify(f"Error: {e}", severity="error")
            raise
    
    def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
        """Handle worker state changes."""
        if event.state == WorkerState.ERROR:
            self.log.error(f"Worker failed: {event.worker.name}")
        elif event.state == WorkerState.SUCCESS:
            self.log.info(f"Worker completed: {event.worker.name}")

多个 Worker

管理并发 Worker：

class MyApp(App):
    def on_mount(self) -> None:
        # Run multiple workers concurrently
        self.run_worker(self.task_one(), name="task1", group="processing")
        self.run_worker(self.task_two(), name="task2", group="processing")
        self.run_worker(self.task_three(), name="task3", group="processing")
    
    async def task_one(self) -> None:
        await asyncio.sleep(2)
        self.notify("Task 1 complete")
    
    async def task_two(self) -> None:
        await asyncio.sleep(3)
        self.notify("Task 2 complete")
    
    async def task_three(self) -> None:
        await asyncio.sleep(1)
        self.notify("Task 3 complete")
    
    def cancel_all_tasks(self) -> None:
        """Cancel all workers in a group."""
        for worker in self.workers:
            if worker.group == "processing":
                worker.cancel()

线程 Worker 与进程 Worker

选择合适的 Worker 类型：

class MyApp(App):
    def on_mount(self) -> None:
        # Async task (default) - for I/O bound operations
        self.run_worker(self.fetch_data())
        
        # Thread worker - for CPU-bound tasks
        self.run_worker(self.process_data(), thread=True)
    
    async def fetch_data(self) -> str:
        """I/O bound: use async."""
        async with httpx.AsyncClient() as client:
            response = await client.get("https://api.example.com")
            return response.text
    
    def process_data(self) -> str:
        """CPU bound: use thread."""
        # Heavy computation
        result = [i**2 for i in range(1000000)]
        return str(sum(result))

Worker 最佳实践

1. 以下情况务必使用 Worker：

   • 网络请求
   • 文件 I/O
   • 数据库查询
   • CPU 密集型计算
   • 任何耗时超过 100 ms 的操作

2. Worker 模式：

   • 使用 exclusive=True 防止重复启动 Worker
   • 为 Worker 命名，以便更轻松地调试
   • 将相关 Worker 分组，以便批量取消
   • 始终处理 Worker 错误

3. 线程安全：

   • 在线程中更新 UI 时使用 call_from_thread()
   • 切勿直接在线程中修改组件
   • 使用锁保护共享的可变状态

4. 取消：

   • 需要取消时保存 Worker 引用
   • 在长循环中检查 worker.is_cancelled
   • 在 finally 代码块中清理资源

模态对话框

from textual.screen import ModalScreen

class ConfirmDialog(ModalScreen[bool]):
    """Modal confirmation dialog."""
    
    def compose(self) -> ComposeResult:
        with Container(id="dialog"):
            yield Label("Are you sure?")
            with Horizontal():
                yield Button("Yes", variant="primary", id="yes")
                yield Button("No", variant="error", id="no")
    
    def on_button_pressed(self, event: Button.Pressed) -> None:
        self.dismiss(event.button.id == "yes")

# Use in app
async def confirm_action(self) -> None:
    result = await self.push_screen_wait(ConfirmDialog())
    if result:
        self.log("Confirmed!")

屏幕与导航

from textual.screen import Screen

class MainScreen(Screen):
    def compose(self) -> ComposeResult:
        yield Header()
        yield Button("Go to Settings")
        yield Footer()
    
    def on_button_pressed(self) -> None:
        self.app.push_screen("settings")

class SettingsScreen(Screen):
    def compose(self) -> ComposeResult:
        yield Label("Settings")
        yield Button("Back")
    
    def on_button_pressed(self) -> None:
        self.app.pop_screen()

class MyApp(App):
    SCREENS = {
        "main": MainScreen(),
        "settings": SettingsScreen(),
    }

测试

使用 pytest 和 Pilot API 测试 Textual 应用：

import pytest
from textual.pilot import Pilot
from my_app import MyApp

@pytest.mark.asyncio
async def test_app_starts():
    app = MyApp()
    async with app.run_test() as pilot:
        assert app.screen is not None

@pytest.mark.asyncio
async def test_button_click():
    app = MyApp()
    async with app.run_test() as pilot:
        await pilot.click("#my-button")
        # Assert expected state changes
        
@pytest.mark.asyncio
async def test_keyboard_input():
    app = MyApp()
    async with app.run_test() as pilot:
        await pilot.press("q")
        # Verify app exited or state changed

最佳实践

性能

• 对按需加载的高开销组件使用 Lazy
• 编写高效的 render() 方法，避免不必要的工作
• 谨慎使用响应式属性，仅用于真正动态的值
• 处理多个更改时批量更新 UI

状态管理

• 将应用状态保存在 App 实例中，以便全局访问
• 对绑定到 UI 的状态使用响应式属性
• 将复杂状态存储在专用的数据模型中
• 避免组件间进行深层嵌套通信

错误处理

from textual.widgets import RichLog

def compose(self) -> ComposeResult:
    yield RichLog(id="log")

async def action_risky_operation(self) -> None:
    try:
        result = await some_async_operation()
        self.notify("Success!", severity="information")
    except Exception as e:
        self.notify(f"Error: {e}", severity="error")
        self.query_one(RichLog).write(f"[red]Error:[/] {e}")

无障碍访问

• 始终提供键盘导航
• 使用语义化的组件名称和 ID
• 在适当情况下加入类似 ARIA 的描述
• 测试时考虑与屏幕阅读器的兼容性

开发工具

Textual 控制台

调试正在运行的应用：

# Terminal 1: Run console
textual console

# Terminal 2: Run app with console enabled
textual run --dev app.py

用于启用控制台的应用代码：

self.log("Debug message")  # Appears in console
self.log.info("Info level")
self.log.error("Error level")

Textual 开发工具

使用开发工具进行实时检查：

pip install textual-dev
textual run --dev app.py  # Enables hot reload

参考资料

• Widget Gallery：请参阅 references/widgets.md，了解全面的组件示例
• Layout Patterns：请参阅 references/layouts.md，了解常见的布局方案
• Styling Guide：请参阅 references/styling.md，了解 CSS 样式模式和主题
• Official Guides Index：请参阅 references/official-guides-index.md，查看所有官方 Textual 文档指南的 URL（需要时使用 web_fetch 获取详细信息）
• 示例应用：请参阅 assets/，查看完整的示例应用

常见问题

1. 忘记使用 async/await：Textual 的许多方法都是异步的，务必使用 await
2. 阻塞事件循环：重要 — 对于长时间运行的任务（网络、I/O、繁重计算），请使用 run_worker()。切勿在主线程中使用 time.sleep() 或执行阻塞操作
3. 错误的消息处理：方法名称必须符合 on_{message_name} 模式
4. CSS 优先级问题：恰当使用 ID 和类，以便有针对性地设置样式
5. 未使用查询方法：使用 query_one() 和 query()，而非手动遍历
6. 违反线程安全规则：切勿直接从工作线程修改组件 — 请使用 call_from_thread()
7. 未取消工作任务：即使屏幕关闭，工作任务仍会继续运行 — 务必取消任务或保存其引用
8. 在异步代码中使用 time.sleep：在异步函数中使用 await asyncio.sleep()，而非 time.sleep()
9. 未处理工作任务错误：工作任务可能会静默失败 — 务必实现错误处理
10. 工作任务类型错误：I/O 操作请使用异步工作任务，CPU 密集型任务请使用线程工作任务