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.

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.