alihan/textual
1
0
Fork 0
mirror of https://github.com/Textualize/textual.git synced 2025-10-17 02:38:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

unified events and messages

Browse Source
This commit is contained in:
Will McGugan
2022-08-10 13:09:38 +01:00
parent d6079deb90
commit b3eb543e38
23 changed files with 137 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/examples/light_dark.py
View File

@@ -5,17 +5,15 @@ from textual.widgets import Button
class ButtonApp(App):
CSS = """
Button {
width: 100%;
}
"""
def compose(self):
yield Button("Lights off")
def handle_pressed(self, event):
def on_button_pressed(self, event):
self.dark = not self.dark
self.bell()
event.button.label = "Lights ON" if self.dark else "Lights OFF"

2
sandbox/buttons.py
View File

@@ -12,7 +12,7 @@ class ButtonsApp(App[str]):
Button.error("error", id="baz"),
)
def handle_pressed(self, event: Button.Pressed) -> None:
def on_button_pressed(self, event: Button.Pressed) -> None:
self.app.bell()
async def on_key(self, event: events.Key) -> None:

2
sandbox/darren/buttons.py
View File

@@ -12,7 +12,7 @@ class ButtonsApp(App[str]):
Button.error("error", id="baz"),
)
def handle_pressed(self, event: Button.Pressed) -> None:
def on_button_pressed(self, event: Button.Pressed) -> None:
self.app.bell()
async def on_key(self, event: events.Key) -> None:

2
sandbox/darren/file_search.py
View File

@@ -58,7 +58,7 @@ class FileSearchApp(App):
self.mount(file_table_wrapper=Widget(self.file_table))
self.mount(search_bar=self.search_bar)
def handle_changed(self, event: TextWidgetBase.Changed) -> None:
def on_text_widget_base_changed(self, event: TextWidgetBase.Changed) -> None:
self.file_table.filter = event.value

2
sandbox/input.py
View File

@@ -48,7 +48,7 @@ class InputApp(App[str]):
)
self.mount(text_area=TextArea())
def handle_changed(self, event: TextWidgetBase.Changed) -> None:
def on_text_widget_base_changed(self, event: TextWidgetBase.Changed) -> None:
try:
value = float(event.value)
except ValueError:

2
sandbox/will/add_remove.py
View File

@@ -46,7 +46,7 @@ class AddRemoveApp(App):
layout.Vertical(id="items"),
)
def handle_pressed(self, event: Button.Pressed) -> None:
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "add":
self.count += 1
self.query("#items").first().mount(

4
sandbox/will/scroll.py
View File

@@ -27,7 +27,7 @@ class ButtonsApp(App[str]):
Button("There can be only one"),
)
def handle_pressed(self, event: Button.Pressed) -> None:
def on_button_pressed(self, event: Button.Pressed) -> None:
self.app.bell()
async def on_key(self, event: events.Key) -> None:
@@ -38,7 +38,7 @@ class ButtonsApp(App[str]):
app = ButtonsApp(
log_path="textual.log", css_path="buttons.css", watch_css=True, log_verbosity=2
log_path="textual.log", css_path="buttons.css", watch_css=True, log_verbosity=3
)
if __name__ == "__main__":

1
src/textual/_callback.py
View File

@@ -23,7 +23,6 @@ async def invoke(callback: Callable, *params: object) -> Any:
"""
_rich_traceback_guard = True
parameter_count = count_parameters(callback)
result = callback(*params[:parameter_count])
if isawaitable(result):
result = await result

9
src/textual/app.py
View File

@@ -1035,7 +1035,8 @@ class App(Generic[ReturnType], DOMNode):
# Forward the event to the view
await self.screen.forward_event(event)
elif isinstance(event, events.Paste):
await self.focused.forward_event(event)
if self.focused is not None:
await self.focused.forward_event(event)
else:
await super().on_event(event)
@@ -1111,11 +1112,11 @@ class App(Generic[ReturnType], DOMNode):
return False
return True
async def handle_update(self, message: messages.Update) -> None:
async def on_update(self, message: messages.Update) -> None:
message.stop()
self._paint()
async def handle_layout(self, message: messages.Layout) -> None:
async def on_layout(self, message: messages.Layout) -> None:
message.stop()
self._paint()
@@ -1167,7 +1168,7 @@ class App(Generic[ReturnType], DOMNode):
async def action_toggle_class(self, selector: str, class_name: str) -> None:
self.screen.query(selector).toggle_class(class_name)
def handle_terminal_supports_synchronized_output(
def on_terminal_supports_synchronized_output(
self, message: messages.TerminalSupportsSynchronizedOutput
) -> None:
log("[b green]SynchronizedOutput mode is supported")

2
src/textual/devtools/borders.py
View File

@@ -50,7 +50,7 @@ class BorderApp(App):
self.text = Static(TEXT)
yield self.text
def handle_pressed(self, event):
def on_button_pressed(self, event: Button.Pressed) -> None:
self.text.styles.border = (
event.button.id,
self.stylesheet.variables["primary"],

5
src/textual/events.py
View File

@@ -28,11 +28,6 @@ class Event(Message):
super().__init_subclass__(bubble=bubble, verbosity=verbosity)
class Null(Event, verbosity=3):
def can_replace(self, message: Message) -> bool:
return isinstance(message, Null)
@rich.repr.auto
class Callback(Event, bubble=False, verbosity=3):
def __init__(

24
src/textual/message.py
View File

@@ -20,15 +20,14 @@ class Message:
"_forwarded",
"_no_default_action",
"_stop_propagation",
"__done_event",
"_handler_name",
]
sender: MessageTarget
bubble: ClassVar[bool] = True # Message will bubble to parent
verbosity: ClassVar[int] = 1 # Verbosity (higher the more verbose)
system: ClassVar[
bool
] = False # Message is system related and may not be handled by client code
no_dispatch: ClassVar[bool] = False # Message may not be handled by client code
namespace: ClassVar[str] = "" # Namespace to disambiguate messages
def __init__(self, sender: MessageTarget) -> None:
"""
@@ -43,6 +42,9 @@ class Message:
self._forwarded = False
self._no_default_action = False
self._stop_propagation = False
self._handler_name = (
f"on_{self.namespace}_{self.name}" if self.namespace else f"on_{self.name}"
)
super().__init__()
def __rich_repr__(self) -> rich.repr.Result:
@@ -52,20 +54,28 @@ class Message:
cls,
bubble: bool | None = True,
verbosity: int | None = 1,
system: bool | None = False,
no_dispatch: bool | None = False,
namespace: str | None = None,
) -> None:
super().__init_subclass__()
if bubble is not None:
cls.bubble = bubble
if verbosity is not None:
cls.verbosity = verbosity
if system is not None:
cls.system = system
if no_dispatch is not None:
cls.no_dispatch = no_dispatch
if namespace is not None:
cls.namespace = namespace
@property
def is_forwarded(self) -> bool:
return self._forwarded
@property
def handler_name(self) -> str:
# Property to make it read only
return self._handler_name
def set_forwarded(self) -> None:
"""Mark this event as being forwarded."""
self._forwarded = True

117
src/textual/message_pump.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
from typing import Any
import inspect
from asyncio import CancelledError
from asyncio import PriorityQueue, QueueEmpty, Task
@@ -10,10 +11,12 @@ from weakref import WeakSet
from . import events
from . import log
from .case import camel_to_snake
from ._timer import Timer, TimerCallback
from ._callback import invoke
from ._context import active_app, NoActiveAppError
from .message import Message
from .events import Event
from . import messages
if TYPE_CHECKING:
@@ -51,7 +54,27 @@ class MessagePriority:
return self.priority > other.priority
class MessagePump:
class MessagePumpMeta(type):
"""Metaclass for message pump. This exists to populate an Event class of a Widget with the
parent classes' name.
"""
def __new__(
cls, name: str, bases: tuple[type, ...], class_dict: dict[str, Any], **kwargs
):
namespace = camel_to_snake(name)
isclass = inspect.isclass
for value in class_dict.values():
if isclass(value) and issubclass(value, Message):
if not value.namespace:
value.namespace = namespace
class_obj = super().__new__(cls, name, bases, class_dict, **kwargs)
return class_obj
class MessagePump(metaclass=MessagePumpMeta):
def __init__(self, parent: MessagePump | None = None) -> None:
self._message_queue: PriorityQueue[MessagePriority] = PriorityQueue()
self._parent = parent
@@ -204,9 +227,9 @@ class MessagePump:
message = messages.InvokeLater(self, partial(callback, *args, **kwargs))
self.post_message_no_wait(message)
def handle_invoke_later(self, message: messages.InvokeLater) -> None:
def on_invoke_later(self, message: messages.InvokeLater) -> None:
# Forward InvokeLater message to the Screen
self.app.screen.post_message_no_wait(message)
self.app.screen._invoke_later(message.callback)
def close_messages_no_wait(self) -> None:
"""Request the message queue to exit."""
@@ -290,20 +313,32 @@ class MessagePump:
log("CLOSED", self)
async def dispatch_message(self, message: Message) -> bool | None:
async def dispatch_message(self, message: Message) -> None:
"""Dispatch a message received form the message queue.
Args:
message (Message): A message object
"""
_rich_traceback_guard = True
if message.system:
return False
if isinstance(message, events.Event):
if not isinstance(message, events.Null):
await self.on_event(message)
if message.no_dispatch:
return
# Allow apps to treat events and messages separately
if isinstance(message, Event):
await self.on_event(message)
else:
return await self.on_message(message)
return False
await self.on_message(message)
def _get_dispatch_methods(
self, method_name: str, message: Message
) -> Iterable[tuple[type, Callable[[Message], Awaitable]]]:
"""Gets handlers from the MRO
Args:
method_name (str): Handler method name.
message (Message): Message object.
"""
for cls in self.__class__.__mro__:
if message._no_default_action:
break
@@ -312,47 +347,57 @@ class MessagePump:
yield cls, method.__get__(self, cls)
async def on_event(self, event: events.Event) -> None:
_rich_traceback_guard = True
"""Called to process an event.
for cls, method in self._get_dispatch_methods(f"on_{event.name}", event):
log(
event,
">>>",
self,
f"method=<{cls.__name__}.{method.__func__.__name__}>",
verbosity=event.verbosity,
)
await invoke(method, event)
if event.bubble and self._parent and not event._stop_propagation:
if event.sender == self._parent:
# parent is sender, so we stop propagation after parent
event.stop()
if self.is_parent_active:
await self._parent.post_message(event)
Args:
event (events.Event): An Event object.
"""
await self.on_message(event)
async def on_message(self, message: Message) -> None:
_rich_traceback_guard = True
method_name = f"handle_{message.name}"
method = getattr(self, method_name, None)
"""Called to process a message.
if method is not None:
log(message, ">>>", self, verbosity=message.verbosity)
Args:
message (Message): A Message object.
"""
_rich_traceback_guard = True
handler_name = message._handler_name
# Look through the MRO to find a handler
for cls, method in self._get_dispatch_methods(handler_name, message):
log(
message,
">>>",
self,
f"method=<{cls.__name__}.{handler_name}>",
verbosity=message.verbosity,
)
await invoke(method, message)
# Bubble messages up the DOM (if enabled on the message)
if message.bubble and self._parent and not message._stop_propagation:
if message.sender == self._parent:
# parent is sender, so we stop propagation after parent
message.stop()
if not self._parent._closed and not self._parent._closing:
if self.is_parent_active and not self._parent._closing:
await self._parent.post_message(message)
def check_idle(self):
def check_idle(self) -> None:
"""Prompt the message pump to call idle if the queue is empty."""
if self._message_queue.empty():
self.post_message_no_wait(messages.Prompt(sender=self))
async def post_message(self, message: Message) -> bool:
"""Post a message or an event to this message pump.
Args:
message (Message): A message object.
Returns:
bool: True if the messages was posted successfully, False if the message was not posted
(because the message pump was in the process of closing).
"""
if self._closing or self._closed:
return False
if not self.check_message_enabled(message):
@@ -374,6 +419,7 @@ class MessagePump:
Returns:
bool: True if the messages was processed.
"""
if self._closing or self._closed:
return False
if not self.check_message_enabled(message):
@@ -382,6 +428,7 @@ class MessagePump:
return True
def post_message_no_wait(self, message: Message) -> bool:
if self._closing or self._closed:
return False
if not self.check_message_enabled(message):

3
src/textual/messages.py
View File

@@ -50,6 +50,7 @@ class InvokeLater(Message, verbosity=3):
yield "callback", self.callback
# TODO: This should really be an Event
@rich.repr.auto
class CursorMove(Message):
def __init__(self, sender: MessagePump, line: int) -> None:
@@ -66,7 +67,7 @@ class StylesUpdated(Message):
return isinstance(message, StylesUpdated)
class Prompt(Message, system=True):
class Prompt(Message, no_dispatch=True):
"""Used to 'wake up' an event loop."""
def can_replace(self, message: Message) -> bool:

16
src/textual/screen.py
View File

@@ -156,9 +156,15 @@ class Screen(Widget):
for callback in callbacks:
await invoke(callback)
def handle_invoke_later(self, message: messages.InvokeLater) -> None:
# Enqueue the callback function to be called later
self._callbacks.append(message.callback)
def _invoke_later(self, callback: CallbackType) -> None:
"""Enqueue a callback to be invoked after the screen is repainted.
Args:
callback (CallbackType): A callback.
"""
self._callbacks.append(callback)
self.check_idle()
def _refresh_layout(self, size: Size | None = None, full: bool = False) -> None:
"""Refresh the layout (can change size and positions of widgets)."""
@@ -201,7 +207,7 @@ class Screen(Widget):
if display_update is not None:
self.app._display(display_update)
async def handle_update(self, message: messages.Update) -> None:
async def on_update(self, message: messages.Update) -> None:
message.stop()
message.prevent_default()
widget = message.widget
@@ -209,7 +215,7 @@ class Screen(Widget):
self._dirty_widgets.add(widget)
self.check_idle()
async def handle_layout(self, message: messages.Layout) -> None:
async def on_layout(self, message: messages.Layout) -> None:
message.stop()
message.prevent_default()
self._layout_required = True

3
src/textual/views/__init__.py
View File

@@ -1,3 +0,0 @@
from ._dock_view import DockView, Dock, DockEdge
from ._grid_view import GridView
from ._window_view import WindowView

72
src/textual/views/_dock_view.py
View File

@@ -1,72 +0,0 @@
from __future__ import annotations
from typing import cast, Optional
from ..layouts.dock import DockLayout, Dock, DockEdge
from ..layouts.grid import GridLayout, GridAlign
from ..screen import Screen
from ..widget import Widget
class DoNotSet:
pass
do_not_set = DoNotSet()
class DockView(Screen):
def __init__(self, name: str | None = None) -> None:
super().__init__(layout=DockLayout(), name=name)
async def dock(
self,
*widgets: Widget,
name: str | None = None,
id: str | None = None,
edge: DockEdge = "top",
z: int = 0,
size: int | None | DoNotSet = do_not_set,
) -> None:
dock = Dock(edge, widgets, z)
assert isinstance(self.layout, DockLayout)
self.layout.docks.append(dock)
for widget in widgets:
if id is not None:
widget._id = id
if name is not None:
widget.name = name
if size is not do_not_set:
widget.layout_size = cast(Optional[int], size)
if name is None:
await self.mount(widget)
else:
await self.mount(**{name: widget})
await self._refresh_layout()
async def dock_grid(
self,
*,
name: str | None = None,
id: str | None = None,
edge: DockEdge = "top",
z: int = 0,
size: int | None | DoNotSet = do_not_set,
gap: tuple[int, int] | int | None = None,
gutter: tuple[int, int] | int | None = None,
align: tuple[GridAlign, GridAlign] | None = None,
) -> GridLayout:
grid = GridLayout(gap=gap, gutter=gutter, align=align)
view = Screen(layout=grid, id=id, name=name)
dock = Dock(edge, (view,), z)
assert isinstance(self.layout, DockLayout)
self.layout.docks.append(dock)
if size is not do_not_set:
view.layout_size = cast(Optional[int], size)
if name is None:
await self.mount(view)
else:
await self.mount(**{name: view})
await self._refresh_layout()
return grid

3
src/textual/views/_document_view.py
View File

@@ -1,3 +0,0 @@
from __future__ import annotations
from ..screen import Screen

9
src/textual/views/_grid_view.py
View File

@@ -1,9 +0,0 @@
from ..screen import Screen
from ..layouts.grid import GridLayout
class GridView(Screen, layout=GridLayout):
@property
def grid(self) -> GridLayout:
assert isinstance(self.layout, GridLayout)
return self.layout

65
src/textual/views/_window_view.py
View File

@@ -1,65 +0,0 @@
from __future__ import annotations
from rich.console import RenderableType
from .. import events
from ..geometry import Size, SpacingDimensions
from ..layouts.vertical import VerticalLayout
from ..screen import Screen
from ..message import Message
from .. import messages
from ..widget import Widget
from ..widgets import Static
class WindowChange(Message):
def can_replace(self, message: Message) -> bool:
return isinstance(message, WindowChange)
class WindowView(Screen, layout=VerticalLayout):
def __init__(
self,
widget: RenderableType | Widget,
*,
auto_width: bool = False,
gutter: SpacingDimensions = (0, 0),
name: str | None = None,
) -> None:
layout = VerticalLayout(gutter=gutter, auto_width=auto_width)
self.widget = widget if isinstance(widget, Widget) else Static(widget)
layout.add(self.widget)
super().__init__(name=name, layout=layout)
async def update(self, widget: Widget | RenderableType) -> None:
layout = self.layout
assert isinstance(layout, VerticalLayout)
layout.clear()
self.widget = widget if isinstance(widget, Widget) else Static(widget)
layout.add(self.widget)
layout.require_update()
self.refresh(layout=True)
await self.emit(WindowChange(self))
async def handle_update(self, message: messages.Update) -> None:
message.prevent_default()
await self.emit(WindowChange(self))
async def handle_layout(self, message: messages.Layout) -> None:
self.layout.require_update()
message.stop()
self.refresh()
async def watch_virtual_size(self, size: Size) -> None:
await self.emit(WindowChange(self))
async def watch_scroll_x(self, value: int) -> None:
self.layout.require_update()
self.refresh()
async def watch_scroll_y(self, value: int) -> None:
self.layout.require_update()
self.refresh()
async def on_resize(self, event: events.Resize) -> None:
await self.emit(WindowChange(self))

15
src/textual/widget.py
View File

@@ -1105,13 +1105,12 @@ class Widget(DOMNode):
Args:
event (events.Idle): Idle event.
"""
if self._repaint_required:
self._repaint_required = False
self.screen.post_message_no_wait(messages.Update(self, self))
if self._layout_required:
self._layout_required = False
self.screen.post_message_no_wait(messages.Layout(self))
self._layout_required = False
self._repaint_required = False
def focus(self) -> None:
"""Give input focus to this widget."""
@@ -1205,27 +1204,27 @@ class Widget(DOMNode):
if self.scroll_up(animate=False):
event.stop()
def handle_scroll_to(self, message: ScrollTo) -> None:
def on_scroll_to(self, message: ScrollTo) -> None:
if self.is_scrollable:
self.scroll_to(message.x, message.y, animate=message.animate, duration=0.1)
message.stop()
def handle_scroll_up(self, event: ScrollUp) -> None:
def on_scroll_up(self, event: ScrollUp) -> None:
if self.is_scrollable:
self.scroll_page_up()
event.stop()
def handle_scroll_down(self, event: ScrollDown) -> None:
def on_scroll_down(self, event: ScrollDown) -> None:
if self.is_scrollable:
self.scroll_page_down()
event.stop()
def handle_scroll_left(self, event: ScrollLeft) -> None:
def on_scroll_left(self, event: ScrollLeft) -> None:
if self.is_scrollable:
self.scroll_page_left()
event.stop()
def handle_scroll_right(self, event: ScrollRight) -> None:
def on_scroll_right(self, event: ScrollRight) -> None:
if self.is_scrollable:
self.scroll_page_right()
event.stop()

3
src/textual/widgets/_button.py
View File

@@ -300,3 +300,6 @@ class Button(Widget, can_focus=True):
id=id,
classes=classes,
)
print(Button.Pressed.namespace)

2
src/textual/widgets/_data_table.py
View File

@@ -220,7 +220,7 @@ class DataTable(ScrollView, Generic[CellType], can_focus=True):
return self.header_height
return self.rows[row_index].height
async def handle_styles_updated(self, message: messages.StylesUpdated) -> None:
async def on_styles_updated(self, message: messages.StylesUpdated) -> None:
self._clear_caches()
self.refresh()