Skip to content

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