textual-reactive-programming（文本 React 式编程）

Textual Reactive Programming

Purpose

Implement efficient, declarative data binding in Textual widgets using reactive attributes. This pattern eliminates manual refresh calls and ensures UI stays synchronized with state.

Quick Start

from textual.reactive import reactive
from textual.widgets import Static

class CounterWidget(Static):
    """Widget with reactive counter."""

    count = reactive(0)  # Reactive attribute initialized to 0

    def render(self) -> str:
        """Auto-called when count changes."""
        return f"Count: {self.count}"

    def increment(self) -> None:
        """Increment counter - triggers render()."""
        self.count += 1

# Usage:
widget = CounterWidget()
widget.increment()  # Automatically re-renders

Instructions

Step 1: Define Reactive Attributes

Declare reactive attributes using reactive():

from textual.reactive import reactive
from textual.widgets import Static
from typing import Literal

class ReactiveWidget(Static):
    """Widget with multiple reactive attributes."""

    # Basic reactive attribute
    status: reactive[str] = reactive("idle", init=False)

    # Reactive with initial value
    count: reactive[int] = reactive(0)

    # Reactive with complex type
    items: reactive[list[str]] = reactive(list, init=False)

    # Reactive with default name (attribute name used)
    message: reactive[str] = reactive("default message")

    # Reactive with type union
    state: reactive[Literal["on", "off"]] = reactive("off")

    def __init__(self, **kwargs: object) -> None:
        super().__init__(**kwargs)
        # Initialize reactive attributes in __init__
        # Only if init=False is set above
        # Otherwise they're initialized by reactive()

Reactive Declaration Options:

reactive(
    default_value,              # Required: initial value
    init=False,                 # True: initialize in __init__, False: initialize here
    layout=False,               # True: trigger layout when changed
    recompose=False,            # True: call compose() when changed
)

Important Rules:

• Type hints are required: attribute: reactive[Type]
• init=False means initialize in __init__
• init=True (default) means initialize by reactive()
• Changing a reactive attribute triggers a watcher/refresh

Step 2: Use Watch Methods

Watch methods automatically trigger when reactive attributes change:

from textual.reactive import reactive
from textual.widgets import Static
from typing import Literal

class WatcherWidget(Static):
    """Widget demonstrating watch methods."""

    status: reactive[Literal["idle", "running", "error"]] = reactive("idle", init=False)
    count: reactive[int] = reactive(0)
    items: reactive[list[str]] = reactive(list, init=False)

    def __init__(self, **kwargs: object) -> None:
        super().__init__(**kwargs)
        self.status = "idle"
        self.items = []

    def watch_status(self, old_value: str, new_value: str) -> None:
        """Called when status changes.

        Args:
            old_value: Previous status.
            new_value: New status.
        """
        # Update UI based on status change
        print(f"Status changed: {old_value} -> {new_value}")
        self.refresh()  # Re-render widget

    def watch_count(self, old_value: int, new_value: int) -> None:
        """Called when count changes."""
        if new_value < 0:
            # Revert invalid change
            self.count = old_value
        else:
            # Count is valid
            self.refresh()

    def watch_items(self, old_value: list[str], new_value: list[str]) -> None:
        """Called when items list changes.

        Note: Called on replacement, not on list mutations.
        """
        print(f"Items changed: {len(old_value)} -> {len(new_value)} items")
        self.refresh()

    def add_item(self, item: str) -> None:
        """Add item (triggers watch_items)."""
        # Replace list to trigger watcher
        self.items = self.items + [item]  # Creates new list
        # NOT: self.items.append(item)  # This won't trigger watcher

    def increment(self) -> None:
        """Increment count (triggers watch_count)."""
        self.count += 1  # Triggers watch_count

Watcher Pattern:

• Method name: watch_{attribute_name}
• Signature: watch_method(self, old_value: Type, new_value: Type) -> None
• Called automatically when attribute changes
• Called BEFORE re-render
• Use for validation, side effects, cascading updates

Step 3: Implement Computed Properties

Use computed attributes that derive from other reactive attributes:

from textual.reactive import reactive, var
from textual.widgets import Static

class ComputedWidget(Static):
    """Widget with computed reactive attributes."""

    first_name: reactive[str] = reactive("John", init=False)
    last_name: reactive[str] = reactive("Doe", init=False)

    # Computed property (read-only)
    @property
    def full_name(self) -> str:
        """Derived from first_name and last_name."""
        return f"{self.first_name} {self.last_name}"

    # Alternative: Use var() for derived reactive attribute
    full_name_reactive: reactive[str] = reactive("John Doe")

    def watch_first_name(self, old_value: str, new_value: str) -> None:
        """Update computed attribute when first_name changes."""
        self.full_name_reactive = f"{new_value} {self.last_name}"
        self.refresh()

    def watch_last_name(self, old_value: str, new_value: str) -> None:
        """Update computed attribute when last_name changes."""
        self.full_name_reactive = f"{self.first_name} {new_value}"
        self.refresh()

    def render(self) -> str:
        return f"Full Name: {self.full_name}"

Pattern:

• Use @property for computed values derived from reactive attrs
• Watchers update derived attributes when dependencies change
• Re-render when computed attributes change

Step 4: Handle Complex State Changes

Manage complex updates with multiple reactive attributes:

from textual.reactive import reactive
from textual.widgets import Static
from dataclasses import dataclass

@dataclass(frozen=True)
class DataPoint:
    """Immutable data point."""
    value: float
    timestamp: str

class ComplexStateWidget(Static):
    """Widget managing complex reactive state."""

    # Multiple reactive attributes
    data: reactive[list[DataPoint]] = reactive(list, init=False)
    total: reactive[float] = reactive(0.0)
    average: reactive[float] = reactive(0.0)
    loading: reactive[bool] = reactive(False)
    error: reactive[str | None] = reactive(None)

    def __init__(self, **kwargs: object) -> None:
        super().__init__(**kwargs)
        self.data = []
        self.total = 0.0
        self.average = 0.0
        self.loading = False
        self.error = None

    def watch_data(self, old_value: list[DataPoint], new_value: list[DataPoint]) -> None:
        """Recalculate totals when data changes."""
        if not new_value:
            self.total = 0.0
            self.average = 0.0
            return

        # Update computed values
        self.total = sum(p.value for p in new_value)
        self.average = self.total / len(new_value)
        self.error = None

    async def load_data(self) -> None:
        """Load data with loading state."""
        try:
            self.loading = True
            self.error = None

            # Simulate async fetch
            import asyncio
            await asyncio.sleep(1)

            # Update data (triggers watch_data)
            self.data = [
                DataPoint(10.5, "2024-01-01"),
                DataPoint(20.3, "2024-01-02"),
                DataPoint(15.8, "2024-01-03"),
            ]
        except Exception as e:
            self.error = str(e)
            self.data = []
        finally:
            self.loading = False

    def render(self) -> str:
        """Render with state."""
        if self.loading:
            return "Loading..."

        if self.error:
            return f"Error: {self.error}"

        if not self.data:
            return "No data"

        return (
            f"Items: {len(self.data)}\n"
            f"Total: {self.total:.2f}\n"
            f"Average: {self.average:.2f}"
        )

Pattern:

• Organize related reactive attributes together
• Watch methods maintain invariants (total, average, etc.)
• Atomic updates: replace entire list/dict to trigger watchers
• Use frozen dataclasses for immutable data points

Step 5: Implement Validation with Reactive Attributes

Use watchers to validate and enforce constraints:

from textual.reactive import reactive
from textual.widgets import Input, Static

class ValidatingWidget(Static):
    """Widget with reactive validation."""

    value: reactive[str] = reactive("", init=False)
    is_valid: reactive[bool] = reactive(True)
    error_message: reactive[str] = reactive("")

    def __init__(self, **kwargs: object) -> None:
        super().__init__(**kwargs)
        self.value = ""
        self.is_valid = True
        self.error_message = ""

    def watch_value(self, old_value: str, new_value: str) -> None:
        """Validate value when it changes."""
        # Validate
        is_valid = self._validate(new_value)

        if is_valid:
            self.is_valid = True
            self.error_message = ""
        else:
            self.is_valid = False
            self.error_message = "Invalid value"

        self.refresh()

    def _validate(self, value: str) -> bool:
        """Validate value according to rules."""
        # Example: email validation
        return "@" in value and "." in value

    def render(self) -> str:
        """Render with validation status."""
        status = "✓" if self.is_valid else "✗"
        return f"{status} {self.value}\n{self.error_message}"

Validation Pattern:

• Watch method validates on change
• Update separate reactive attributes for validity
• Re-render to show status
• Use color changes via CSS for invalid state

Examples

Example 1: Agent Status Widget with Reactive Updates

from textual.reactive import reactive
from textual.widgets import Static
from rich.text import Text
from typing import Literal

class AgentStatusWidget(Static):
    """Reactive agent status widget."""

    agent_id: reactive[str] = reactive("")
    status: reactive[Literal["idle", "running", "error"]] = reactive("idle", init=False)
    task_count: reactive[int] = reactive(0)
    error_message: reactive[str | None] = reactive(None)

    DEFAULT_CSS = """
    AgentStatusWidget {
        height: auto;
        border: solid $primary;
        padding: 1;
    }

    AgentStatusWidget .status-idle {
        color: $warning;
    }

    AgentStatusWidget .status-running {
        color: $success;
    }

    AgentStatusWidget .status-error {
        color: $error;
    }
    """

    def __init__(self, agent_id: str, **kwargs: object) -> None:
        super().__init__(**kwargs)
        self.agent_id = agent_id
        self.status = "idle"
        self.task_count = 0
        self.error_message = None

    def watch_status(self, old_value: str, new_value: str) -> None:
        """Update display when status changes."""
        self.refresh()

    def watch_task_count(self, old_value: int, new_value: int) -> None:
        """Track task count changes."""
        if new_value > old_value:
            # Task started
            if self.status == "idle":
                self.status = "running"
        elif new_value == 0:
            # All tasks completed
            if self.status == "running":
                self.status = "idle"

    def watch_error_message(self, old_value: str | None, new_value: str | None) -> None:
        """Update status when error occurs."""
        if new_value:
            self.status = "error"
        else:
            self.status = "idle"

    def render(self) -> Text:
        """Render reactive agent status."""
        text = Text()

        # Status indicator
        icon = self._get_icon()
        text.append(icon, style=f"status-{self.status}")
        text.append(" ")

        # Agent info
        text.append(self.agent_id, style="bold")
        text.append(f" ({self.status.upper()})")

        # Task count
        if self.task_count > 0:
            text.append(f" [{self.task_count} tasks]", style="dim")

        # Error
        if self.error_message:
            text.append("\n")
            text.append(f"Error: {self.error_message}", style="bold red")

        return text

    def _get_icon(self) -> str:
        """Get status icon."""
        return {
            "idle": "◐",
            "running": "●",
            "error": "✗",
        }.get(self.status, "?")

    async def update_task_count(self, count: int) -> None:
        """Update task count."""
        self.task_count = count

    async def set_error(self, error: str | None) -> None:
        """Set error state."""
        self.error_message = error

Example 2: Form Widget with Reactive Validation

from textual.reactive import reactive
from textual.app import ComposeResult
from textual.containers import Container
from textual.widgets import Static, Input, Button

class FormWidget(Container):
    """Form with reactive validation."""

    email: reactive[str] = reactive("", init=False)
    password: reactive[str] = reactive("", init=False)
    is_valid: reactive[bool] = reactive(False)

    DEFAULT_CSS = """
    FormWidget {
        height: auto;
        border: solid $primary;
        padding: 1;
    }

    FormWidget .field {
        height: auto;
        margin: 0 0 1 0;
    }

    FormWidget .field-label {
        text-style: bold;
    }

    FormWidget .field-error {
        color: $error;
        text-style: dim;
    }

    FormWidget .submit-button {
        margin-top: 1;
    }
    """

    def __init__(self, **kwargs: object) -> None:
        super().__init__(**kwargs)
        self.email = ""
        self.password = ""
        self.is_valid = False

    def compose(self) -> ComposeResult:
        yield Static("Email", classes="field-label")
        yield Input(id="email-input", classes="field")
        yield Static("", id="email-error", classes="field-error")

        yield Static("Password", classes="field-label")
        yield Input(id="password-input", password=True, classes="field")
        yield Static("", id="password-error", classes="field-error")

        yield Button("Submit", id="submit", disabled=True, classes="submit-button")

    async def on_mount(self) -> None:
        """Set up input watchers."""
        email_input = self.query_one("#email-input", Input)
        password_input = self.query_one("#password-input", Input)

        email_input.on_change_handler = self._on_email_change
        password_input.on_change_handler = self._on_password_change

    async def _on_email_change(self, value: str) -> None:
        """Handle email input change."""
        self.email = value
        await self._validate_email()

    async def _on_password_change(self, value: str) -> None:
        """Handle password input change."""
        self.password = value
        await self._validate_password()

    def watch_email(self, old_value: str, new_value: str) -> None:
        """Validate email when changed."""
        self._check_form_valid()

    def watch_password(self, old_value: str, new_value: str) -> None:
        """Validate password when changed."""
        self._check_form_valid()

    async def _validate_email(self) -> None:
        """Validate email format."""
        error_widget = self.query_one("#email-error", Static)
        if not self.email or "@" not in self.email:
            error_widget.update("Invalid email")
        else:
            error_widget.update("")

    async def _validate_password(self) -> None:
        """Validate password strength."""
        error_widget = self.query_one("#password-error", Static)
        if len(self.password) < 8:
            error_widget.update("Password must be 8+ characters")
        else:
            error_widget.update("")

    def _check_form_valid(self) -> None:
        """Check if entire form is valid."""
        is_valid = (
            "@" in self.email
            and len(self.password) >= 8
        )
        self.is_valid = is_valid

        # Update submit button
        submit = self.query_one("#submit", Button)
        submit.disabled = not is_valid

Requirements

• Textual >= 0.45.0
• Python 3.9+ with type hints

Best Practices

1. Prefer reactive attributes over manual refresh:

# ❌ WRONG - manual refresh
def set_status(self, status: str) -> None:
    self._status = status
    self.refresh()

# ✅ CORRECT - reactive attribute
status: reactive[str] = reactive("")

def watch_status(self, old_value: str, new_value: str) -> None:
    """Auto-refresh on change."""
    pass  # refresh() called automatically

2. Use watchers for side effects:

def watch_count(self, old_value: int, new_value: int) -> None:
    """Handle side effects."""
    if new_value > 10:
        self.error = "Too many items"
    self.refresh()

3. Replace collections to trigger watchers:

# ❌ WRONG - mutation doesn't trigger watcher
self.items.append(new_item)

# ✅ CORRECT - replacement triggers watcher
self.items = self.items + [new_item]
# OR
self.items = [*self.items, new_item]

See Also

• textual-widget-development.md - Widget structure and lifecycle
• textual-event-messages.md - Custom messages and events