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.

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

Name	Type	Description	Default
callback	Callback	A callable.	required

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

Name	Type	Description	Default
callback	Callback	Callable to call next.	required
*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

Name	Type	Description	Default
callback	Callback	Callable to run after current event.	required
*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

Name	Type	Description	Default
message	Message	A message object.	required

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

Name	Type	Description	Default
event	events.Key	A key event.	required

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

Name	Type	Description	Default
event	events.Event	An Event object.	required

post_message method

def post_message(self, message):

Posts a message on to this widget's queue.

Parameters

Name	Type	Description	Default
message	Message	A message (including Event).	required

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

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

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

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

Returns

Type	Description
Timer	A timer object.