textual-tui

Textual TUI 開発

インタラクティブな TUI アプリケーションを作成する、最新の Python フレームワークである Textual を使用して、実運用に適した品質のターミナルユーザーインターフェースを構築します。

クイックスタート

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")

ワーカーとバックグラウンドタスク

重要：UI のブロックを防ぐため、時間のかかる処理には必ずワーカーを使用してください。イベントループは応答可能な状態を保つ必要があります。

基本的なワーカーの使用方法

バックグラウンドスレッドでタスクを実行します：

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"

進捗状況を更新するワーカー

処理中に 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)

ワーカーの通信パターン

スレッドセーフな UI 更新には call_from_thread を使用します：

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)

ワーカーのキャンセル

不要になったワーカーをキャンセルします：

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

ワーカーのエラー処理

ワーカーの失敗を適切に処理します：

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}")

複数のワーカー

並行するワーカーを管理します：

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()

スレッドワーカーとプロセスワーカー

適切なワーカーの種類を選択します：

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))

ワーカーのベストプラクティス

1. 以下の場合は必ずワーカーを使用します：

   • ネットワークリクエスト
   • ファイル I/O
   • データベースクエリ
   • CPU 負荷の高い計算
   • 100ms を超える処理

2. ワーカーのパターン：

   • 重複するワーカーを防ぐには exclusive=True を使用します
   • デバッグしやすいようにワーカーに名前を付けます
   • 関連するワーカーをグループ化し、一括キャンセルできるようにします
   • ワーカーのエラーは必ず処理します

3. スレッドセーフティ：

   • スレッドから UI を更新するには call_from_thread() を使用します
   • スレッドからウィジェットを直接変更しないでください
   • 共有される可変状態にはロックを使用します

4. キャンセル：

   • キャンセルが必要な場合は、ワーカーへの参照を保持します
   • 長いループでは 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

参考資料

• ウィジェットギャラリー：包括的なウィジェットの例は、references/widgets.md を参照してください
• レイアウトパターン：一般的なレイアウトのレシピは、references/layouts.md を参照してください
• スタイルガイド：CSS のパターンとテーマは、references/styling.md を参照してください
• 公式ガイドの索引：すべての公式 Textual ドキュメントガイドの URL は、references/official-guides-index.md を参照してください（詳細情報は必要に応じて 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 の使用：非同期関数では time.sleep() ではなく、await asyncio.sleep() を使用してください
9. ワーカーエラーへの未対応：ワーカーはエラーを通知せずに失敗することがあります。必ずエラー処理を実装してください
10. 誤ったワーカータイプの使用：I/O には非同期ワーカー、CPU 負荷の高いタスクにはスレッドワーカーを使用してください