Message pump

A MessagePump is a base class for any object which processes messages, which includes Widget, Screen, and App.

Tip

Most of the method here are useful in general app development.

MessagePump class

def __init__(self, parent=None):

Base class which supplies a message pump.

app property

app: 'App[object]'

Get the current app.

Returns

Type	Description
'App[object]'	The current app.

Raises

Type	Description
NoActiveAppError	if no active app could be found for the current asyncio context

has_parent property

has_parent: bool

Does this object have a parent?

is_attached property

is_attached: bool

Is the node attached to the app via the DOM?

is_parent_active property

is_parent_active: bool

Is the parent active?

is_running property

is_running: bool

Is the message pump running (potentially processing messages)?

log property

log: Logger

Get a logger for this object.

Returns

Type	Description
Logger	A logger.

message_queue_size property

message_queue_size: int

The current size of the message queue.

call_after_refresh method

def call_after_refresh(self, callback, *args, **kwargs):

Schedule a callback to run after all messages are processed and the screen has been refreshed. Positional and keyword arguments are passed to the callable.

Parameters

Parameter	Default	Description
callback Callback	required	A callable.

Returns

Type	Description
bool	True if the callback was scheduled, or False if the callback could not be scheduled (may occur if the message pump was closed or closing).

call_later method

def call_later(self, callback, *args, **kwargs):

Schedule a callback to run after all messages are processed in this object. Positional and keywords arguments are passed to the callable.

Parameters

Parameter	Default	Description
callback Callback	required	Callable to call next.
*args Any	()	Positional arguments to pass to the callable.
**kwargs Any	{}	Keyword arguments to pass to the callable.

Returns

Type	Description
bool	True if the callback was scheduled, or False if the callback could not be scheduled (may occur if the message pump was closed or closing).

call_next method

def call_next(self, callback, *args, **kwargs):

Schedule a callback to run immediately after processing the current message.

Parameters

Parameter	Default	Description
callback Callback	required	Callable to run after current event.
*args Any	()	Positional arguments to pass to the callable.
**kwargs Any	{}	Keyword arguments to pass to the callable.

check_idle method

def check_idle(self):

Prompt the message pump to call idle if the queue is empty.

check_message_enabled method

def check_message_enabled(self, message):

Check if a given message is enabled (allowed to be sent).

Parameters

Parameter	Default	Description
message Message	required	A message object.

Returns

Type	Description
bool	True if the message will be sent, or False if it is disabled.

disable_messages method

def disable_messages(self, *messages):

Disable message types from being processed.

dispatch_key async

def dispatch_key(self, event):

Dispatch a key event to method.

This method will call the method named 'key_' if it exists. Some keys have aliases. The first alias found will be invoked if it exists. If multiple handlers exist that match the key, an exception is raised.

Parameters

Parameter	Default	Description
event events.Key	required	A key event.

Returns

Type	Description
bool	True if key was handled, otherwise False.

Raises

Type	Description
DuplicateKeyHandlers	When there's more than 1 handler that could handle this key.

enable_messages method

def enable_messages(self, *messages):

Enable processing of messages types.

on_event async

def on_event(self, event):

Called to process an event.

Parameters

Parameter	Default	Description
event events.Event	required	An Event object.

post_message method

def post_message(self, message):

Posts a message on to this widget's queue.

Parameters

Parameter	Default	Description
message Message	required	A message (including Event).

Returns

Type	Description
bool	True if the messages was processed, False if it wasn't.

prevent method

def prevent(self, *message_types):

A context manager to temporarily prevent the given message types from being posted.

Example

input = self.query_one(Input)
with self.prevent(Input.Changed):
    input.value = "foo"

set_interval method

def set_interval(
    self,
    interval,
    callback=None,
    *,
    name=None,
    repeat=0,
    pause=False
):

Call a function at periodic intervals.

Parameters

Parameter	Default	Description
interval float	required	Time (in seconds) between calls.
callback TimerCallback | None	None	Function to call.
name str | None	None	Name of the timer object.
repeat int	0	Number of times to repeat the call or 0 for continuous.
pause bool	False	Start the timer paused.

Returns

Type	Description
Timer	A timer object.

set_timer method

def set_timer(
    self, delay, callback=None, *, name=None, pause=False
):

Make a function call after a delay.

Parameters

Parameter	Default	Description
delay float	required	Time (in seconds) to wait before invoking callback.
callback TimerCallback | None	None	Callback to call after time has expired.
name str | None	None	Name of the timer (for debug).
pause bool	False	Start timer paused.

Returns

Type	Description
Timer	A timer object.