Zalgorithm

Textual TUI framework events and messages

Events are a kind of message sent by Textual in response to input and other state changes.1

Docs: https://textual.textualize.io/guide/events/

Every App and Widget object has a message queue. Messages are pulled of the queue and processed by a handler method.

Messages are processed with an asyncio task that’s started when the widget (or the app) is mounted. The asynio task monitors its queue for new messages and dispatches them to the appropriate handler.

The Textual docs are good. For now I’ll just note a few things that I want to remember.

Message handlers #

Message handlers

Most of the logic in Textual apps will be in the message handler functions. Handlers are named with on_ + the message’s namespace + _ + the name of the message class. E.g., on_button_selected.

The on decorator #

Instead of the on_messagenamespace_messageclass handler naming convention, the on decorator can be used:

@on(Button.Pressed)
def handle_button_pressed(self):
# ...

A benefit of the on decorator approach is that you can specify which widgets the message should be handled for.

Instead of the following (copied from my actual (messy) code):

@on(Input.Submitted)
def input_changed(self, event: Input.Submitted) -> None:
    if event.input.id == "r-a":
        self.a.r = float(event.input.value)
        self.osc("/complex/a", self.a.r, self.a.theta)
    if event.input.id == "theta-a":
        self.a.theta = float(event.input.value) * np.pi / 180
        self.osc("/complex/a", self.a.r, self.a.theta)
    if event.input.id == "r-b":
        self.b.r = float(event.input.value)
        self.osc("/complex/b", self.b.r, self.b.theta)
    if event.input.id == "theta-b":
        self.b.theta = float(event.input.value) * np.pi / 180
        self.osc("/complex/b", self.b.r, self.b.theta)

I could have written:

@on(Input.Submitted, "#r-a")
def handle_real_a_submitted(self, event: Input.Submitted) -> None:
    self.a.r = float(event.input.value)
    self.osc("/complex/a", self.a.r, self.a.theta)

Async message handlers #

I’m starting to work through this in notes / A conceptual overview of asyncio

Message/event handlers that are prefixed with the async keyword will be coroutines.

Notes related to coroutines:

From the Python docs: https://docs.python.org/3/library/asyncio-task.html

If handlers are prefixed with async, Textual will await them. This lets the handler use the await keyword for anynchronous APIs.

My understanding of concurrency gets fuzzy here:

The solution is to launch a new asyncio task to do the network task in the background.

The asyncio-task documentation has a clear example: https://docs.python.org/3/library/asyncio-task.html#coroutines .

EDIT: there are some details in notes / Running coroutines in Python that will make things more clear. The key is in what the Textual docs are saying: “Message handlers may be coroutines…bear in mind that an individual widget will not be able to pick up a new message from its message queue until the handler has returned…the solution is to launch a new asyncio task to do the network task in the background.”

Here’s the example that’s used for creating a task:

    async def on_input_changed(self, message: Input.Changed) -> None:
        """A coroutine to handle a text changed message."""
        if message.value:
            # Look up the word in the background
            # *** The important part:
            asyncio.create_task(self.lookup_word(message.value))
        else:
            # Clear the results
            self.query_one("#results", Static).update()

References #

textual.textualize. “Events and Messages.” Accessed on: February 2, 2026. https://textual.textualize.io/guide/events/ .


  1. textual.textualize, “Events and Messages,” Accessed on: February 2, 2026, https://textual.textualize.io/guide/events/↩︎