Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import asyncio
4import contextvars
5import os
6import re
7import signal
8import sys
9import threading
10import time
11from asyncio import (
12 AbstractEventLoop,
13 Future,
14 Task,
15 ensure_future,
16 get_running_loop,
17 sleep,
18)
19from collections.abc import Callable, Coroutine, Generator, Hashable, Iterable, Iterator
20from contextlib import ExitStack, contextmanager
21from subprocess import Popen
22from traceback import format_tb
23from typing import (
24 Any,
25 Generic,
26 Literal,
27 TypeVar,
28 cast,
29 overload,
30)
32from prompt_toolkit.buffer import Buffer
33from prompt_toolkit.cache import SimpleCache
34from prompt_toolkit.clipboard import Clipboard, InMemoryClipboard
35from prompt_toolkit.cursor_shapes import AnyCursorShapeConfig, to_cursor_shape_config
36from prompt_toolkit.data_structures import Size
37from prompt_toolkit.enums import EditingMode
38from prompt_toolkit.eventloop import (
39 InputHook,
40 get_traceback_from_context,
41 new_eventloop_with_inputhook,
42 run_in_executor_with_context,
43)
44from prompt_toolkit.eventloop.utils import call_soon_threadsafe
45from prompt_toolkit.filters import Condition, Filter, FilterOrBool, to_filter
46from prompt_toolkit.formatted_text import AnyFormattedText
47from prompt_toolkit.input.base import Input
48from prompt_toolkit.input.typeahead import get_typeahead, store_typeahead
49from prompt_toolkit.key_binding.bindings.page_navigation import (
50 load_page_navigation_bindings,
51)
52from prompt_toolkit.key_binding.defaults import load_key_bindings
53from prompt_toolkit.key_binding.emacs_state import EmacsState
54from prompt_toolkit.key_binding.key_bindings import (
55 Binding,
56 ConditionalKeyBindings,
57 GlobalOnlyKeyBindings,
58 KeyBindings,
59 KeyBindingsBase,
60 KeysTuple,
61 merge_key_bindings,
62)
63from prompt_toolkit.key_binding.key_processor import KeyPressEvent, KeyProcessor
64from prompt_toolkit.key_binding.vi_state import ViState
65from prompt_toolkit.keys import Keys
66from prompt_toolkit.layout.containers import Container, Window
67from prompt_toolkit.layout.controls import BufferControl, UIControl
68from prompt_toolkit.layout.dummy import create_dummy_layout
69from prompt_toolkit.layout.layout import Layout, walk
70from prompt_toolkit.output import ColorDepth, Output
71from prompt_toolkit.renderer import Renderer, print_formatted_text
72from prompt_toolkit.search import SearchState
73from prompt_toolkit.styles import (
74 BaseStyle,
75 DummyStyle,
76 DummyStyleTransformation,
77 DynamicStyle,
78 StyleTransformation,
79 default_pygments_style,
80 default_ui_style,
81 merge_styles,
82)
83from prompt_toolkit.utils import Event, in_main_thread
85from .current import get_app_session, set_app
86from .run_in_terminal import in_terminal, run_in_terminal
88__all__ = [
89 "Application",
90]
93E = KeyPressEvent
94_AppResult = TypeVar("_AppResult")
95ApplicationEventHandler = Callable[["Application[_AppResult]"], None]
97_SIGWINCH = getattr(signal, "SIGWINCH", None)
98_SIGTSTP = getattr(signal, "SIGTSTP", None)
101class Application(Generic[_AppResult]):
102 """
103 The main Application class!
104 This glues everything together.
106 :param layout: A :class:`~prompt_toolkit.layout.Layout` instance.
107 :param key_bindings:
108 :class:`~prompt_toolkit.key_binding.KeyBindingsBase` instance for
109 the key bindings.
110 :param clipboard: :class:`~prompt_toolkit.clipboard.Clipboard` to use.
111 :param full_screen: When True, run the application on the alternate screen buffer.
112 :param color_depth: Any :class:`~.ColorDepth` value, a callable that
113 returns a :class:`~.ColorDepth` or `None` for default.
114 :param erase_when_done: (bool) Clear the application output when it finishes.
115 :param reverse_vi_search_direction: Normally, in Vi mode, a '/' searches
116 forward and a '?' searches backward. In Readline mode, this is usually
117 reversed.
118 :param min_redraw_interval: Number of seconds to wait between redraws. Use
119 this for applications where `invalidate` is called a lot. This could cause
120 a lot of terminal output, which some terminals are not able to process.
122 `None` means that every `invalidate` will be scheduled right away
123 (which is usually fine).
125 When one `invalidate` is called, but a scheduled redraw of a previous
126 `invalidate` call has not been executed yet, nothing will happen in any
127 case.
129 :param max_render_postpone_time: When there is high CPU (a lot of other
130 scheduled calls), postpone the rendering max x seconds. '0' means:
131 don't postpone. '.5' means: try to draw at least twice a second.
133 :param refresh_interval: Automatically invalidate the UI every so many
134 seconds. When `None` (the default), only invalidate when `invalidate`
135 has been called.
137 :param terminal_size_polling_interval: Poll the terminal size every so many
138 seconds. Useful if the applications runs in a thread other then then
139 main thread where SIGWINCH can't be handled, or on Windows.
141 Filters:
143 :param mouse_support: (:class:`~prompt_toolkit.filters.Filter` or
144 boolean). When True, enable mouse support.
145 :param paste_mode: :class:`~prompt_toolkit.filters.Filter` or boolean.
146 :param editing_mode: :class:`~prompt_toolkit.enums.EditingMode`.
148 :param enable_page_navigation_bindings: When `True`, enable the page
149 navigation key bindings. These include both Emacs and Vi bindings like
150 page-up, page-down and so on to scroll through pages. Mostly useful for
151 creating an editor or other full screen applications. Probably, you
152 don't want this for the implementation of a REPL. By default, this is
153 enabled if `full_screen` is set.
155 Callbacks (all of these should accept an
156 :class:`~prompt_toolkit.application.Application` object as input.)
158 :param on_reset: Called during reset.
159 :param on_invalidate: Called when the UI has been invalidated.
160 :param before_render: Called right before rendering.
161 :param after_render: Called right after rendering.
163 I/O:
164 (Note that the preferred way to change the input/output is by creating an
165 `AppSession` with the required input/output objects. If you need multiple
166 applications running at the same time, you have to create a separate
167 `AppSession` using a `with create_app_session():` block.
169 :param input: :class:`~prompt_toolkit.input.Input` instance.
170 :param output: :class:`~prompt_toolkit.output.Output` instance. (Probably
171 Vt100_Output or Win32Output.)
173 Usage::
175 app = Application(...)
176 app.run()
178 # Or
179 await app.run_async()
180 """
182 def __init__(
183 self,
184 layout: Layout | None = None,
185 style: BaseStyle | None = None,
186 include_default_pygments_style: FilterOrBool = True,
187 style_transformation: StyleTransformation | None = None,
188 key_bindings: KeyBindingsBase | None = None,
189 clipboard: Clipboard | None = None,
190 full_screen: bool = False,
191 color_depth: (ColorDepth | Callable[[], ColorDepth | None] | None) = None,
192 mouse_support: FilterOrBool = False,
193 enable_page_navigation_bindings: None
194 | (FilterOrBool) = None, # Can be None, True or False.
195 paste_mode: FilterOrBool = False,
196 editing_mode: EditingMode = EditingMode.EMACS,
197 erase_when_done: bool = False,
198 reverse_vi_search_direction: FilterOrBool = False,
199 min_redraw_interval: float | int | None = None,
200 max_render_postpone_time: float | int | None = 0.01,
201 refresh_interval: float | None = None,
202 terminal_size_polling_interval: float | None = 0.5,
203 cursor: AnyCursorShapeConfig = None,
204 on_reset: ApplicationEventHandler[_AppResult] | None = None,
205 on_invalidate: ApplicationEventHandler[_AppResult] | None = None,
206 before_render: ApplicationEventHandler[_AppResult] | None = None,
207 after_render: ApplicationEventHandler[_AppResult] | None = None,
208 # I/O.
209 input: Input | None = None,
210 output: Output | None = None,
211 ) -> None:
212 # If `enable_page_navigation_bindings` is not specified, enable it in
213 # case of full screen applications only. This can be overridden by the user.
214 if enable_page_navigation_bindings is None:
215 enable_page_navigation_bindings = Condition(lambda: self.full_screen)
217 paste_mode = to_filter(paste_mode)
218 mouse_support = to_filter(mouse_support)
219 reverse_vi_search_direction = to_filter(reverse_vi_search_direction)
220 enable_page_navigation_bindings = to_filter(enable_page_navigation_bindings)
221 include_default_pygments_style = to_filter(include_default_pygments_style)
223 if layout is None:
224 layout = create_dummy_layout()
226 if style_transformation is None:
227 style_transformation = DummyStyleTransformation()
229 self.style = style
230 self.style_transformation = style_transformation
232 # Key bindings.
233 self.key_bindings = key_bindings
234 self._default_bindings = load_key_bindings()
235 self._page_navigation_bindings = load_page_navigation_bindings()
237 self.layout = layout
238 self.clipboard = clipboard or InMemoryClipboard()
239 self.full_screen: bool = full_screen
240 self._color_depth = color_depth
241 self.mouse_support = mouse_support
243 self.paste_mode = paste_mode
244 self.editing_mode = editing_mode
245 self.erase_when_done = erase_when_done
246 self.reverse_vi_search_direction = reverse_vi_search_direction
247 self.enable_page_navigation_bindings = enable_page_navigation_bindings
248 self.min_redraw_interval = min_redraw_interval
249 self.max_render_postpone_time = max_render_postpone_time
250 self.refresh_interval = refresh_interval
251 self.terminal_size_polling_interval = terminal_size_polling_interval
253 self.cursor = to_cursor_shape_config(cursor)
255 # Events.
256 self.on_invalidate = Event(self, on_invalidate)
257 self.on_reset = Event(self, on_reset)
258 self.before_render = Event(self, before_render)
259 self.after_render = Event(self, after_render)
261 # I/O.
262 session = get_app_session()
263 self.output = output or session.output
264 self.input = input or session.input
266 # List of 'extra' functions to execute before a Application.run.
267 self.pre_run_callables: list[Callable[[], None]] = []
269 self._is_running = False
270 self.future: Future[_AppResult] | None = None
271 self.loop: AbstractEventLoop | None = None
272 self._loop_thread: threading.Thread | None = None
273 self.context: contextvars.Context | None = None
275 #: Quoted insert. This flag is set if we go into quoted insert mode.
276 self.quoted_insert = False
278 #: Vi state. (For Vi key bindings.)
279 self.vi_state = ViState()
280 self.emacs_state = EmacsState()
282 #: When to flush the input (For flushing escape keys.) This is important
283 #: on terminals that use vt100 input. We can't distinguish the escape
284 #: key from for instance the left-arrow key, if we don't know what follows
285 #: after "\x1b". This little timer will consider "\x1b" to be escape if
286 #: nothing did follow in this time span.
287 #: This seems to work like the `ttimeoutlen` option in Vim.
288 self.ttimeoutlen = 0.5 # Seconds.
290 #: Like Vim's `timeoutlen` option. This can be `None` or a float. For
291 #: instance, suppose that we have a key binding AB and a second key
292 #: binding A. If the uses presses A and then waits, we don't handle
293 #: this binding yet (unless it was marked 'eager'), because we don't
294 #: know what will follow. This timeout is the maximum amount of time
295 #: that we wait until we call the handlers anyway. Pass `None` to
296 #: disable this timeout.
297 self.timeoutlen = 1.0
299 #: The `Renderer` instance.
300 # Make sure that the same stdout is used, when a custom renderer has been passed.
301 self._merged_style = self._create_merged_style(include_default_pygments_style)
303 self.renderer = Renderer(
304 self._merged_style,
305 self.output,
306 full_screen=full_screen,
307 mouse_support=mouse_support,
308 cpr_not_supported_callback=self.cpr_not_supported_callback,
309 )
311 #: Render counter. This one is increased every time the UI is rendered.
312 #: It can be used as a key for caching certain information during one
313 #: rendering.
314 self.render_counter = 0
316 # Invalidate flag. When 'True', a repaint has been scheduled.
317 self._invalidated = False
318 self._invalidate_events: list[
319 Event[object]
320 ] = [] # Collection of 'invalidate' Event objects.
321 self._last_redraw_time = 0.0 # Unix timestamp of last redraw. Used when
322 # `min_redraw_interval` is given.
324 #: The `InputProcessor` instance.
325 self.key_processor = KeyProcessor(_CombinedRegistry(self))
327 # If `run_in_terminal` was called. This will point to a `Future` what will be
328 # set at the point when the previous run finishes.
329 self._running_in_terminal = False
330 self._running_in_terminal_f: Future[None] | None = None
332 # Trigger initialize callback.
333 self.reset()
335 def _create_merged_style(self, include_default_pygments_style: Filter) -> BaseStyle:
336 """
337 Create a `Style` object that merges the default UI style, the default
338 pygments style, and the custom user style.
339 """
340 dummy_style = DummyStyle()
341 pygments_style = default_pygments_style()
343 @DynamicStyle
344 def conditional_pygments_style() -> BaseStyle:
345 if include_default_pygments_style():
346 return pygments_style
347 else:
348 return dummy_style
350 return merge_styles(
351 [
352 default_ui_style(),
353 conditional_pygments_style,
354 DynamicStyle(lambda: self.style),
355 ]
356 )
358 @property
359 def color_depth(self) -> ColorDepth:
360 """
361 The active :class:`.ColorDepth`.
363 The current value is determined as follows:
365 - If a color depth was given explicitly to this application, use that
366 value.
367 - Otherwise, fall back to the color depth that is reported by the
368 :class:`.Output` implementation. If the :class:`.Output` class was
369 created using `output.defaults.create_output`, then this value is
370 coming from the $PROMPT_TOOLKIT_COLOR_DEPTH environment variable.
371 """
372 depth = self._color_depth
374 if callable(depth):
375 depth = depth()
377 if depth is None:
378 depth = self.output.get_default_color_depth()
380 return depth
382 @property
383 def current_buffer(self) -> Buffer:
384 """
385 The currently focused :class:`~.Buffer`.
387 (This returns a dummy :class:`.Buffer` when none of the actual buffers
388 has the focus. In this case, it's really not practical to check for
389 `None` values or catch exceptions every time.)
390 """
391 return self.layout.current_buffer or Buffer(
392 name="dummy-buffer"
393 ) # Dummy buffer.
395 @property
396 def current_search_state(self) -> SearchState:
397 """
398 Return the current :class:`.SearchState`. (The one for the focused
399 :class:`.BufferControl`.)
400 """
401 ui_control = self.layout.current_control
402 if isinstance(ui_control, BufferControl):
403 return ui_control.search_state
404 else:
405 return SearchState() # Dummy search state. (Don't return None!)
407 def reset(self) -> None:
408 """
409 Reset everything, for reading the next input.
410 """
411 # Notice that we don't reset the buffers. (This happens just before
412 # returning, and when we have multiple buffers, we clearly want the
413 # content in the other buffers to remain unchanged between several
414 # calls of `run`. (And the same is true for the focus stack.)
416 self.exit_style = ""
418 self._background_tasks: set[Task[None]] = set()
420 self.renderer.reset()
421 self.key_processor.reset()
422 self.layout.reset()
423 self.vi_state.reset()
424 self.emacs_state.reset()
426 # Trigger reset event.
427 self.on_reset.fire()
429 # Make sure that we have a 'focusable' widget focused.
430 # (The `Layout` class can't determine this.)
431 layout = self.layout
433 if not layout.current_control.is_focusable():
434 for w in layout.find_all_windows():
435 if w.content.is_focusable():
436 layout.current_window = w
437 break
439 def invalidate(self) -> None:
440 """
441 Thread safe way of sending a repaint trigger to the input event loop.
442 """
443 if not self._is_running:
444 # Don't schedule a redraw if we're not running.
445 # Otherwise, `get_running_loop()` in `call_soon_threadsafe` can fail.
446 # See: https://github.com/dbcli/mycli/issues/797
447 return
449 # `invalidate()` called if we don't have a loop yet (not running?), or
450 # after the event loop was closed.
451 if self.loop is None or self.loop.is_closed():
452 return
454 # Never schedule a second redraw, when a previous one has not yet been
455 # executed. (This should protect against other threads calling
456 # 'invalidate' many times, resulting in 100% CPU.)
457 if self._invalidated:
458 return
459 else:
460 self._invalidated = True
462 # Trigger event.
463 self.loop.call_soon_threadsafe(self.on_invalidate.fire)
465 def redraw() -> None:
466 self._invalidated = False
467 self._redraw()
469 def schedule_redraw() -> None:
470 call_soon_threadsafe(
471 redraw, max_postpone_time=self.max_render_postpone_time, loop=self.loop
472 )
474 if self.min_redraw_interval:
475 # When a minimum redraw interval is set, wait minimum this amount
476 # of time between redraws.
477 diff = time.time() - self._last_redraw_time
478 if diff < self.min_redraw_interval:
480 async def redraw_in_future() -> None:
481 await sleep(cast(float, self.min_redraw_interval) - diff)
482 schedule_redraw()
484 self.loop.call_soon_threadsafe(
485 lambda: self.create_background_task(redraw_in_future())
486 )
487 else:
488 schedule_redraw()
489 else:
490 schedule_redraw()
492 @property
493 def invalidated(self) -> bool:
494 "True when a redraw operation has been scheduled."
495 return self._invalidated
497 def _redraw(self, render_as_done: bool = False) -> None:
498 """
499 Render the command line again. (Not thread safe!) (From other threads,
500 or if unsure, use :meth:`.Application.invalidate`.)
502 :param render_as_done: make sure to put the cursor after the UI.
503 """
505 def run_in_context() -> None:
506 # Only draw when no sub application was started.
507 if self._is_running and not self._running_in_terminal:
508 if self.min_redraw_interval:
509 self._last_redraw_time = time.time()
511 # Render
512 self.render_counter += 1
513 self.before_render.fire()
515 if render_as_done:
516 if self.erase_when_done:
517 self.renderer.erase()
518 else:
519 # Draw in 'done' state and reset renderer.
520 self.renderer.render(self, self.layout, is_done=render_as_done)
521 else:
522 self.renderer.render(self, self.layout)
524 self.layout.update_parents_relations()
526 # Fire render event.
527 self.after_render.fire()
529 self._update_invalidate_events()
531 # NOTE: We want to make sure this Application is the active one. The
532 # invalidate function is often called from a context where this
533 # application is not the active one. (Like the
534 # `PromptSession._auto_refresh_context`).
535 # We copy the context in case the context was already active, to
536 # prevent RuntimeErrors. (The rendering is not supposed to change
537 # any context variables.)
538 if self.context is not None:
539 self.context.copy().run(run_in_context)
541 def _start_auto_refresh_task(self) -> None:
542 """
543 Start a while/true loop in the background for automatic invalidation of
544 the UI.
545 """
546 if self.refresh_interval is not None and self.refresh_interval != 0:
548 async def auto_refresh(refresh_interval: float) -> None:
549 while True:
550 await sleep(refresh_interval)
551 self.invalidate()
553 self.create_background_task(auto_refresh(self.refresh_interval))
555 def _update_invalidate_events(self) -> None:
556 """
557 Make sure to attach 'invalidate' handlers to all invalidate events in
558 the UI.
559 """
560 # Remove all the original event handlers. (Components can be removed
561 # from the UI.)
562 for ev in self._invalidate_events:
563 ev -= self._invalidate_handler
565 # Gather all new events.
566 # (All controls are able to invalidate themselves.)
567 def gather_events() -> Iterable[Event[object]]:
568 for c in self.layout.find_all_controls():
569 yield from c.get_invalidate_events()
571 self._invalidate_events = list(gather_events())
573 for ev in self._invalidate_events:
574 ev += self._invalidate_handler
576 def _invalidate_handler(self, sender: object) -> None:
577 """
578 Handler for invalidate events coming from UIControls.
580 (This handles the difference in signature between event handler and
581 `self.invalidate`. It also needs to be a method -not a nested
582 function-, so that we can remove it again .)
583 """
584 self.invalidate()
586 def _on_resize(self) -> None:
587 """
588 When the window size changes, we erase the current output and request
589 again the cursor position. When the CPR answer arrives, the output is
590 drawn again.
591 """
592 # Erase, request position (when cursor is at the start position)
593 # and redraw again. -- The order is important.
594 self.renderer.erase(leave_alternate_screen=False)
595 self._request_absolute_cursor_position()
596 self._redraw()
598 def _pre_run(self, pre_run: Callable[[], None] | None = None) -> None:
599 """
600 Called during `run`.
602 `self.future` should be set to the new future at the point where this
603 is called in order to avoid data races. `pre_run` can be used to set a
604 `threading.Event` to synchronize with UI termination code, running in
605 another thread that would call `Application.exit`. (See the progress
606 bar code for an example.)
607 """
608 if pre_run:
609 pre_run()
611 # Process registered "pre_run_callables" and clear list.
612 for c in self.pre_run_callables:
613 c()
614 del self.pre_run_callables[:]
616 async def run_async(
617 self,
618 pre_run: Callable[[], None] | None = None,
619 set_exception_handler: bool = True,
620 handle_sigint: bool = True,
621 slow_callback_duration: float = 0.5,
622 ) -> _AppResult:
623 """
624 Run the prompt_toolkit :class:`~prompt_toolkit.application.Application`
625 until :meth:`~prompt_toolkit.application.Application.exit` has been
626 called. Return the value that was passed to
627 :meth:`~prompt_toolkit.application.Application.exit`.
629 This is the main entry point for a prompt_toolkit
630 :class:`~prompt_toolkit.application.Application` and usually the only
631 place where the event loop is actually running.
633 :param pre_run: Optional callable, which is called right after the
634 "reset" of the application.
635 :param set_exception_handler: When set, in case of an exception, go out
636 of the alternate screen and hide the application, display the
637 exception, and wait for the user to press ENTER.
638 :param handle_sigint: Handle SIGINT signal if possible. This will call
639 the `<sigint>` key binding when a SIGINT is received. (This only
640 works in the main thread.)
641 :param slow_callback_duration: Display warnings if code scheduled in
642 the asyncio event loop takes more time than this. The asyncio
643 default of `0.1` is sometimes not sufficient on a slow system,
644 because exceptionally, the drawing of the app, which happens in the
645 event loop, can take a bit longer from time to time.
646 """
647 assert not self._is_running, "Application is already running."
649 if not in_main_thread() or sys.platform == "win32":
650 # Handling signals in other threads is not supported.
651 # Also on Windows, `add_signal_handler(signal.SIGINT, ...)` raises
652 # `NotImplementedError`.
653 # See: https://github.com/prompt-toolkit/python-prompt-toolkit/issues/1553
654 handle_sigint = False
656 async def _run_async(f: asyncio.Future[_AppResult]) -> _AppResult:
657 context = contextvars.copy_context()
658 self.context = context
660 # Counter for cancelling 'flush' timeouts. Every time when a key is
661 # pressed, we start a 'flush' timer for flushing our escape key. But
662 # when any subsequent input is received, a new timer is started and
663 # the current timer will be ignored.
664 flush_task: asyncio.Task[None] | None = None
666 # Reset.
667 # (`self.future` needs to be set when `pre_run` is called.)
668 self.reset()
669 self._pre_run(pre_run)
671 # Feed type ahead input first.
672 self.key_processor.feed_multiple(get_typeahead(self.input))
673 self.key_processor.process_keys()
675 def read_from_input() -> None:
676 nonlocal flush_task
678 # Ignore when we aren't running anymore. This callback will
679 # removed from the loop next time. (It could be that it was
680 # still in the 'tasks' list of the loop.)
681 # Except: if we need to process incoming CPRs.
682 if not self._is_running and not self.renderer.waiting_for_cpr:
683 return
685 # Get keys from the input object.
686 keys = self.input.read_keys()
688 # Feed to key processor.
689 self.key_processor.feed_multiple(keys)
690 self.key_processor.process_keys()
692 # Quit when the input stream was closed.
693 if self.input.closed:
694 if not f.done():
695 f.set_exception(EOFError)
696 else:
697 # Automatically flush keys.
698 if flush_task:
699 flush_task.cancel()
700 flush_task = self.create_background_task(auto_flush_input())
702 def read_from_input_in_context() -> None:
703 # Ensure that key bindings callbacks are always executed in the
704 # current context. This is important when key bindings are
705 # accessing contextvars. (These callbacks are currently being
706 # called from a different context. Underneath,
707 # `loop.add_reader` is used to register the stdin FD.)
708 # (We copy the context to avoid a `RuntimeError` in case the
709 # context is already active.)
710 context.copy().run(read_from_input)
712 async def auto_flush_input() -> None:
713 # Flush input after timeout.
714 # (Used for flushing the enter key.)
715 # This sleep can be cancelled, in that case we won't flush yet.
716 await sleep(self.ttimeoutlen)
717 flush_input()
719 def flush_input() -> None:
720 if not self.is_done:
721 # Get keys, and feed to key processor.
722 keys = self.input.flush_keys()
723 self.key_processor.feed_multiple(keys)
724 self.key_processor.process_keys()
726 if self.input.closed:
727 f.set_exception(EOFError)
729 # Enter raw mode, attach input and attach WINCH event handler.
730 with (
731 self.input.raw_mode(),
732 self.input.attach(read_from_input_in_context),
733 attach_winch_signal_handler(self._on_resize),
734 ):
735 # Draw UI.
736 self._request_absolute_cursor_position()
737 self._redraw()
738 self._start_auto_refresh_task()
740 self.create_background_task(self._poll_output_size())
742 # Wait for UI to finish.
743 try:
744 result = await f
745 finally:
746 # In any case, when the application finishes.
747 # (Successful, or because of an error.)
748 try:
749 self._redraw(render_as_done=True)
750 finally:
751 # _redraw has a good chance to fail if it calls widgets
752 # with bad code. Make sure to reset the renderer
753 # anyway.
754 self.renderer.reset()
756 # Unset `is_running`, this ensures that possibly
757 # scheduled draws won't paint during the following
758 # yield.
759 self._is_running = False
761 # Detach event handlers for invalidate events.
762 # (Important when a UIControl is embedded in multiple
763 # applications, like ptterm in pymux. An invalidate
764 # should not trigger a repaint in terminated
765 # applications.)
766 for ev in self._invalidate_events:
767 ev -= self._invalidate_handler
768 self._invalidate_events = []
770 # Wait for CPR responses.
771 if self.output.responds_to_cpr:
772 await self.renderer.wait_for_cpr_responses()
774 # Wait for the run-in-terminals to terminate.
775 previous_run_in_terminal_f = self._running_in_terminal_f
777 if previous_run_in_terminal_f:
778 await previous_run_in_terminal_f
780 # Store unprocessed input as typeahead for next time.
781 store_typeahead(self.input, self.key_processor.empty_queue())
783 return result
785 @contextmanager
786 def set_loop() -> Iterator[AbstractEventLoop]:
787 loop = get_running_loop()
788 self.loop = loop
789 self._loop_thread = threading.current_thread()
791 try:
792 yield loop
793 finally:
794 self.loop = None
795 self._loop_thread = None
797 @contextmanager
798 def set_is_running() -> Iterator[None]:
799 self._is_running = True
800 try:
801 yield
802 finally:
803 self._is_running = False
805 @contextmanager
806 def set_handle_sigint(loop: AbstractEventLoop) -> Iterator[None]:
807 if handle_sigint:
808 with _restore_sigint_from_ctypes():
809 # save sigint handlers (python and os level)
810 # See: https://github.com/prompt-toolkit/python-prompt-toolkit/issues/1576
811 loop.add_signal_handler(
812 signal.SIGINT,
813 lambda *_: loop.call_soon_threadsafe(
814 self.key_processor.send_sigint
815 ),
816 )
817 try:
818 yield
819 finally:
820 loop.remove_signal_handler(signal.SIGINT)
821 else:
822 yield
824 @contextmanager
825 def set_exception_handler_ctx(loop: AbstractEventLoop) -> Iterator[None]:
826 if set_exception_handler:
827 previous_exc_handler = loop.get_exception_handler()
828 loop.set_exception_handler(self._handle_exception)
829 try:
830 yield
831 finally:
832 loop.set_exception_handler(previous_exc_handler)
834 else:
835 yield
837 @contextmanager
838 def set_callback_duration(loop: AbstractEventLoop) -> Iterator[None]:
839 # Set slow_callback_duration.
840 original_slow_callback_duration = loop.slow_callback_duration
841 loop.slow_callback_duration = slow_callback_duration
842 try:
843 yield
844 finally:
845 # Reset slow_callback_duration.
846 loop.slow_callback_duration = original_slow_callback_duration
848 @contextmanager
849 def create_future(
850 loop: AbstractEventLoop,
851 ) -> Iterator[asyncio.Future[_AppResult]]:
852 f = loop.create_future()
853 self.future = f # XXX: make sure to set this before calling '_redraw'.
855 try:
856 yield f
857 finally:
858 # Also remove the Future again. (This brings the
859 # application back to its initial state, where it also
860 # doesn't have a Future.)
861 self.future = None
863 with ExitStack() as stack:
864 stack.enter_context(set_is_running())
866 # Make sure to set `_invalidated` to `False` to begin with,
867 # otherwise we're not going to paint anything. This can happen if
868 # this application had run before on a different event loop, and a
869 # paint was scheduled using `call_soon_threadsafe` with
870 # `max_postpone_time`.
871 self._invalidated = False
873 loop = stack.enter_context(set_loop())
875 stack.enter_context(set_handle_sigint(loop))
876 stack.enter_context(set_exception_handler_ctx(loop))
877 stack.enter_context(set_callback_duration(loop))
878 stack.enter_context(set_app(self))
879 stack.enter_context(self._enable_breakpointhook())
881 f = stack.enter_context(create_future(loop))
883 try:
884 return await _run_async(f)
885 finally:
886 # Wait for the background tasks to be done. This needs to
887 # go in the finally! If `_run_async` raises
888 # `KeyboardInterrupt`, we still want to wait for the
889 # background tasks.
890 await self.cancel_and_wait_for_background_tasks()
892 # The `ExitStack` above is defined in typeshed in a way that it can
893 # swallow exceptions. Without next line, mypy would think that there's
894 # a possibility we don't return here. See:
895 # https://github.com/python/mypy/issues/7726
896 assert False, "unreachable"
898 def run(
899 self,
900 pre_run: Callable[[], None] | None = None,
901 set_exception_handler: bool = True,
902 handle_sigint: bool = True,
903 in_thread: bool = False,
904 inputhook: InputHook | None = None,
905 ) -> _AppResult:
906 """
907 A blocking 'run' call that waits until the UI is finished.
909 This will run the application in a fresh asyncio event loop.
911 :param pre_run: Optional callable, which is called right after the
912 "reset" of the application.
913 :param set_exception_handler: When set, in case of an exception, go out
914 of the alternate screen and hide the application, display the
915 exception, and wait for the user to press ENTER.
916 :param in_thread: When true, run the application in a background
917 thread, and block the current thread until the application
918 terminates. This is useful if we need to be sure the application
919 won't use the current event loop (asyncio does not support nested
920 event loops). A new event loop will be created in this background
921 thread, and that loop will also be closed when the background
922 thread terminates. When this is used, it's especially important to
923 make sure that all asyncio background tasks are managed through
924 `get_app().create_background_task()`, so that unfinished tasks are
925 properly cancelled before the event loop is closed. This is used
926 for instance in ptpython.
927 :param handle_sigint: Handle SIGINT signal. Call the key binding for
928 `Keys.SIGINT`. (This only works in the main thread.)
929 """
930 if in_thread:
931 result: _AppResult
932 exception: BaseException | None = None
934 def run_in_thread() -> None:
935 nonlocal result, exception
936 try:
937 result = self.run(
938 pre_run=pre_run,
939 set_exception_handler=set_exception_handler,
940 # Signal handling only works in the main thread.
941 handle_sigint=False,
942 inputhook=inputhook,
943 )
944 except BaseException as e:
945 exception = e
947 thread = threading.Thread(target=run_in_thread)
948 thread.start()
949 thread.join()
951 if exception is not None:
952 raise exception
953 return result
955 coro = self.run_async(
956 pre_run=pre_run,
957 set_exception_handler=set_exception_handler,
958 handle_sigint=handle_sigint,
959 )
961 def _called_from_ipython() -> bool:
962 try:
963 return (
964 sys.modules["IPython"].version_info < (8, 18, 0, "")
965 and "IPython/terminal/interactiveshell.py"
966 in sys._getframe(3).f_code.co_filename
967 )
968 except BaseException:
969 return False
971 if inputhook is not None:
972 # Create new event loop with given input hook and run the app.
973 # In Python 3.12, we can use asyncio.run(loop_factory=...)
974 # For now, use `run_until_complete()`.
975 loop = new_eventloop_with_inputhook(inputhook)
976 result = loop.run_until_complete(coro)
977 loop.run_until_complete(loop.shutdown_asyncgens())
978 loop.close()
979 return result
981 elif _called_from_ipython():
982 # workaround to make input hooks work for IPython until
983 # https://github.com/ipython/ipython/pull/14241 is merged.
984 # IPython was setting the input hook by installing an event loop
985 # previously.
986 try:
987 # See whether a loop was installed already. If so, use that.
988 # That's required for the input hooks to work, they are
989 # installed using `set_event_loop`.
990 loop = asyncio.get_event_loop()
991 except RuntimeError:
992 # No loop installed. Run like usual.
993 return asyncio.run(coro)
994 else:
995 # Use existing loop.
996 return loop.run_until_complete(coro)
998 else:
999 # No loop installed. Run like usual.
1000 return asyncio.run(coro)
1002 def _handle_exception(
1003 self, loop: AbstractEventLoop, context: dict[str, Any]
1004 ) -> None:
1005 """
1006 Handler for event loop exceptions.
1007 This will print the exception, using run_in_terminal.
1008 """
1009 # For Python 2: we have to get traceback at this point, because
1010 # we're still in the 'except:' block of the event loop where the
1011 # traceback is still available. Moving this code in the
1012 # 'print_exception' coroutine will loose the exception.
1013 tb = get_traceback_from_context(context)
1014 formatted_tb = "".join(format_tb(tb))
1016 async def in_term() -> None:
1017 async with in_terminal():
1018 # Print output. Similar to 'loop.default_exception_handler',
1019 # but don't use logger. (This works better on Python 2.)
1020 print("\nUnhandled exception in event loop:")
1021 print(formatted_tb)
1022 print("Exception {}".format(context.get("exception")))
1024 await _do_wait_for_enter("Press ENTER to continue...")
1026 ensure_future(in_term())
1028 @contextmanager
1029 def _enable_breakpointhook(self) -> Generator[None, None, None]:
1030 """
1031 Install our custom breakpointhook for the duration of this context
1032 manager. (We will only install the hook if no other custom hook was
1033 set.)
1034 """
1035 if sys.breakpointhook == sys.__breakpointhook__:
1036 sys.breakpointhook = self._breakpointhook
1038 try:
1039 yield
1040 finally:
1041 sys.breakpointhook = sys.__breakpointhook__
1042 else:
1043 yield
1045 def _breakpointhook(self, *a: object, **kw: object) -> None:
1046 """
1047 Breakpointhook which uses PDB, but ensures that the application is
1048 hidden and input echoing is restored during each debugger dispatch.
1050 This can be called from any thread. In any case, the application's
1051 event loop will be blocked while the PDB input is displayed. The event
1052 will continue after leaving the debugger.
1053 """
1054 app = self
1055 # Inline import on purpose. We don't want to import pdb, if not needed.
1056 import pdb
1057 from types import FrameType
1059 TraceDispatch = Callable[
1060 [FrameType, Literal["call", "line", "return", "exception", "opcode"], Any],
1061 Any,
1062 ]
1064 @contextmanager
1065 def hide_app_from_eventloop_thread() -> Generator[None, None, None]:
1066 """Stop application if `__breakpointhook__` is called from within
1067 the App's event loop."""
1068 # Hide application.
1069 app.renderer.erase()
1071 # Detach input and dispatch to debugger.
1072 with app.input.detach():
1073 with app.input.cooked_mode():
1074 yield
1076 # Note: we don't render the application again here, because
1077 # there's a good chance that there's a breakpoint on the next
1078 # line. This paint/erase cycle would move the PDB prompt back
1079 # to the middle of the screen.
1081 @contextmanager
1082 def hide_app_from_other_thread() -> Generator[None, None, None]:
1083 """Stop application if `__breakpointhook__` is called from a
1084 thread other than the App's event loop."""
1085 ready = threading.Event()
1086 done = threading.Event()
1088 async def in_loop() -> None:
1089 # from .run_in_terminal import in_terminal
1090 # async with in_terminal():
1091 # ready.set()
1092 # await asyncio.get_running_loop().run_in_executor(None, done.wait)
1093 # return
1095 # Hide application.
1096 app.renderer.erase()
1098 # Detach input and dispatch to debugger.
1099 with app.input.detach():
1100 with app.input.cooked_mode():
1101 ready.set()
1102 # Here we block the App's event loop thread until the
1103 # debugger resumes. We could have used `with
1104 # run_in_terminal.in_terminal():` like the commented
1105 # code above, but it seems to work better if we
1106 # completely stop the main event loop while debugging.
1107 done.wait()
1109 self.create_background_task(in_loop())
1110 ready.wait()
1111 try:
1112 yield
1113 finally:
1114 done.set()
1116 class CustomPdb(pdb.Pdb):
1117 def trace_dispatch(
1118 self, frame: FrameType, event: str, arg: Any
1119 ) -> TraceDispatch:
1120 if app._loop_thread is None:
1121 return super().trace_dispatch(frame, event, arg)
1123 if app._loop_thread == threading.current_thread():
1124 with hide_app_from_eventloop_thread():
1125 return super().trace_dispatch(frame, event, arg)
1127 with hide_app_from_other_thread():
1128 return super().trace_dispatch(frame, event, arg)
1130 frame = sys._getframe().f_back
1131 CustomPdb(stdout=sys.__stdout__).set_trace(frame)
1133 def create_background_task(
1134 self, coroutine: Coroutine[Any, Any, None]
1135 ) -> asyncio.Task[None]:
1136 """
1137 Start a background task (coroutine) for the running application. When
1138 the `Application` terminates, unfinished background tasks will be
1139 cancelled.
1141 Given that we still support Python versions before 3.11, we can't use
1142 task groups (and exception groups), because of that, these background
1143 tasks are not allowed to raise exceptions. If they do, we'll call the
1144 default exception handler from the event loop.
1146 If at some point, we have Python 3.11 as the minimum supported Python
1147 version, then we can use a `TaskGroup` (with the lifetime of
1148 `Application.run_async()`, and run run the background tasks in there.
1150 This is not threadsafe.
1151 """
1152 loop = self.loop or get_running_loop()
1153 task: asyncio.Task[None] = loop.create_task(coroutine)
1154 self._background_tasks.add(task)
1156 task.add_done_callback(self._on_background_task_done)
1157 return task
1159 def _on_background_task_done(self, task: asyncio.Task[None]) -> None:
1160 """
1161 Called when a background task completes. Remove it from
1162 `_background_tasks`, and handle exceptions if any.
1163 """
1164 self._background_tasks.discard(task)
1166 if task.cancelled():
1167 return
1169 exc = task.exception()
1170 if exc is not None:
1171 get_running_loop().call_exception_handler(
1172 {
1173 "message": f"prompt_toolkit.Application background task {task!r} "
1174 "raised an unexpected exception.",
1175 "exception": exc,
1176 "task": task,
1177 }
1178 )
1180 async def cancel_and_wait_for_background_tasks(self) -> None:
1181 """
1182 Cancel all background tasks, and wait for the cancellation to complete.
1183 If any of the background tasks raised an exception, this will also
1184 propagate the exception.
1186 (If we had nurseries like Trio, this would be the `__aexit__` of a
1187 nursery.)
1188 """
1189 for task in self._background_tasks:
1190 task.cancel()
1192 # Wait until the cancellation of the background tasks completes.
1193 # `asyncio.wait()` does not propagate exceptions raised within any of
1194 # these tasks, which is what we want. Otherwise, we can't distinguish
1195 # between a `CancelledError` raised in this task because it got
1196 # cancelled, and a `CancelledError` raised on this `await` checkpoint,
1197 # because *we* got cancelled during the teardown of the application.
1198 # (If we get cancelled here, then it's important to not suppress the
1199 # `CancelledError`, and have it propagate.)
1200 # NOTE: Currently, if we get cancelled at this point then we can't wait
1201 # for the cancellation to complete (in the future, we should be
1202 # using anyio or Python's 3.11 TaskGroup.)
1203 # Also, if we had exception groups, we could propagate an
1204 # `ExceptionGroup` if something went wrong here. Right now, we
1205 # don't propagate exceptions, but have them printed in
1206 # `_on_background_task_done`.
1207 if len(self._background_tasks) > 0:
1208 await asyncio.wait(
1209 self._background_tasks, timeout=None, return_when=asyncio.ALL_COMPLETED
1210 )
1212 async def _poll_output_size(self) -> None:
1213 """
1214 Coroutine for polling the terminal dimensions.
1216 Useful for situations where `attach_winch_signal_handler` is not sufficient:
1217 - If we are not running in the main thread.
1218 - On Windows.
1219 """
1220 size: Size | None = None
1221 interval = self.terminal_size_polling_interval
1223 if interval is None:
1224 return
1226 while True:
1227 await asyncio.sleep(interval)
1228 new_size = self.output.get_size()
1230 if size is not None and new_size != size:
1231 self._on_resize()
1232 size = new_size
1234 def cpr_not_supported_callback(self) -> None:
1235 """
1236 Called when we don't receive the cursor position response in time.
1237 """
1238 if not self.output.responds_to_cpr:
1239 return # We know about this already.
1241 def in_terminal() -> None:
1242 self.output.write(
1243 "WARNING: your terminal doesn't support cursor position requests (CPR).\r\n"
1244 )
1245 self.output.flush()
1247 run_in_terminal(in_terminal)
1249 @overload
1250 def exit(self) -> None:
1251 "Exit without arguments."
1253 @overload
1254 def exit(self, *, result: _AppResult, style: str = "") -> None:
1255 "Exit with `_AppResult`."
1257 @overload
1258 def exit(
1259 self, *, exception: BaseException | type[BaseException], style: str = ""
1260 ) -> None:
1261 "Exit with exception."
1263 def exit(
1264 self,
1265 result: _AppResult | None = None,
1266 exception: BaseException | type[BaseException] | None = None,
1267 style: str = "",
1268 ) -> None:
1269 """
1270 Exit application.
1272 .. note::
1274 If `Application.exit` is called before `Application.run()` is
1275 called, then the `Application` won't exit (because the
1276 `Application.future` doesn't correspond to the current run). Use a
1277 `pre_run` hook and an event to synchronize the closing if there's a
1278 chance this can happen.
1280 :param result: Set this result for the application.
1281 :param exception: Set this exception as the result for an application. For
1282 a prompt, this is often `EOFError` or `KeyboardInterrupt`.
1283 :param style: Apply this style on the whole content when quitting,
1284 often this is 'class:exiting' for a prompt. (Used when
1285 `erase_when_done` is not set.)
1286 """
1287 assert result is None or exception is None
1289 if self.future is None:
1290 raise Exception("Application is not running. Application.exit() failed.")
1292 if self.future.done():
1293 raise Exception("Return value already set. Application.exit() failed.")
1295 self.exit_style = style
1297 if exception is not None:
1298 self.future.set_exception(exception)
1299 else:
1300 self.future.set_result(cast(_AppResult, result))
1302 def _request_absolute_cursor_position(self) -> None:
1303 """
1304 Send CPR request.
1305 """
1306 # Note: only do this if the input queue is not empty, and a return
1307 # value has not been set. Otherwise, we won't be able to read the
1308 # response anyway.
1309 if not self.key_processor.input_queue and not self.is_done:
1310 self.renderer.request_absolute_cursor_position()
1312 async def run_system_command(
1313 self,
1314 command: str,
1315 wait_for_enter: bool = True,
1316 display_before_text: AnyFormattedText = "",
1317 wait_text: str = "Press ENTER to continue...",
1318 ) -> None:
1319 """
1320 Run system command (While hiding the prompt. When finished, all the
1321 output will scroll above the prompt.)
1323 :param command: Shell command to be executed.
1324 :param wait_for_enter: FWait for the user to press enter, when the
1325 command is finished.
1326 :param display_before_text: If given, text to be displayed before the
1327 command executes.
1328 :return: A `Future` object.
1329 """
1330 async with in_terminal():
1331 # Try to use the same input/output file descriptors as the one,
1332 # used to run this application.
1333 try:
1334 input_fd = self.input.fileno()
1335 except AttributeError:
1336 input_fd = sys.stdin.fileno()
1337 try:
1338 output_fd = self.output.fileno()
1339 except AttributeError:
1340 output_fd = sys.stdout.fileno()
1342 # Run sub process.
1343 def run_command() -> None:
1344 self.print_text(display_before_text)
1345 p = Popen(command, shell=True, stdin=input_fd, stdout=output_fd)
1346 p.wait()
1348 await run_in_executor_with_context(run_command)
1350 # Wait for the user to press enter.
1351 if wait_for_enter:
1352 await _do_wait_for_enter(wait_text)
1354 def suspend_to_background(self, suspend_group: bool = True) -> None:
1355 """
1356 (Not thread safe -- to be called from inside the key bindings.)
1357 Suspend process.
1359 :param suspend_group: When true, suspend the whole process group.
1360 (This is the default, and probably what you want.)
1361 """
1362 # Only suspend when the operating system supports it.
1363 # (Not on Windows.)
1364 if _SIGTSTP is not None:
1366 def run() -> None:
1367 signal = cast(int, _SIGTSTP)
1368 # Send `SIGTSTP` to own process.
1369 # This will cause it to suspend.
1371 # Usually we want the whole process group to be suspended. This
1372 # handles the case when input is piped from another process.
1373 if suspend_group:
1374 os.kill(0, signal)
1375 else:
1376 os.kill(os.getpid(), signal)
1378 run_in_terminal(run)
1380 def print_text(
1381 self, text: AnyFormattedText, style: BaseStyle | None = None
1382 ) -> None:
1383 """
1384 Print a list of (style_str, text) tuples to the output.
1385 (When the UI is running, this method has to be called through
1386 `run_in_terminal`, otherwise it will destroy the UI.)
1388 :param text: List of ``(style_str, text)`` tuples.
1389 :param style: Style class to use. Defaults to the active style in the CLI.
1390 """
1391 print_formatted_text(
1392 output=self.output,
1393 formatted_text=text,
1394 style=style or self._merged_style,
1395 color_depth=self.color_depth,
1396 style_transformation=self.style_transformation,
1397 )
1399 @property
1400 def is_running(self) -> bool:
1401 "`True` when the application is currently active/running."
1402 return self._is_running
1404 @property
1405 def is_done(self) -> bool:
1406 if self.future:
1407 return self.future.done()
1408 return False
1410 def get_used_style_strings(self) -> list[str]:
1411 """
1412 Return a list of used style strings. This is helpful for debugging, and
1413 for writing a new `Style`.
1414 """
1415 attrs_for_style = self.renderer._attrs_for_style
1417 if attrs_for_style:
1418 return sorted(
1419 re.sub(r"\s+", " ", style_str).strip()
1420 for style_str in attrs_for_style.keys()
1421 )
1423 return []
1426class _CombinedRegistry(KeyBindingsBase):
1427 """
1428 The `KeyBindings` of key bindings for a `Application`.
1429 This merges the global key bindings with the one of the current user
1430 control.
1431 """
1433 def __init__(self, app: Application[_AppResult]) -> None:
1434 self.app = app
1435 self._cache: SimpleCache[
1436 tuple[Window, frozenset[UIControl]], KeyBindingsBase
1437 ] = SimpleCache()
1439 @property
1440 def _version(self) -> Hashable:
1441 """Not needed - this object is not going to be wrapped in another
1442 KeyBindings object."""
1443 raise NotImplementedError
1445 @property
1446 def bindings(self) -> list[Binding]:
1447 """Not needed - this object is not going to be wrapped in another
1448 KeyBindings object."""
1449 raise NotImplementedError
1451 def _create_key_bindings(
1452 self, current_window: Window, other_controls: list[UIControl]
1453 ) -> KeyBindingsBase:
1454 """
1455 Create a `KeyBindings` object that merges the `KeyBindings` from the
1456 `UIControl` with all the parent controls and the global key bindings.
1457 """
1458 key_bindings = []
1459 collected_containers = set()
1461 # Collect key bindings from currently focused control and all parent
1462 # controls. Don't include key bindings of container parent controls.
1463 container: Container = current_window
1464 while True:
1465 collected_containers.add(container)
1466 kb = container.get_key_bindings()
1467 if kb is not None:
1468 key_bindings.append(kb)
1470 if container.is_modal():
1471 break
1473 parent = self.app.layout.get_parent(container)
1474 if parent is None:
1475 break
1476 else:
1477 container = parent
1479 # Include global bindings (starting at the top-model container).
1480 for c in walk(container):
1481 if c not in collected_containers:
1482 kb = c.get_key_bindings()
1483 if kb is not None:
1484 key_bindings.append(GlobalOnlyKeyBindings(kb))
1486 # Add App key bindings
1487 if self.app.key_bindings:
1488 key_bindings.append(self.app.key_bindings)
1490 # Add mouse bindings.
1491 key_bindings.append(
1492 ConditionalKeyBindings(
1493 self.app._page_navigation_bindings,
1494 self.app.enable_page_navigation_bindings,
1495 )
1496 )
1497 key_bindings.append(self.app._default_bindings)
1499 # Reverse this list. The current control's key bindings should come
1500 # last. They need priority.
1501 key_bindings = key_bindings[::-1]
1503 return merge_key_bindings(key_bindings)
1505 @property
1506 def _key_bindings(self) -> KeyBindingsBase:
1507 current_window = self.app.layout.current_window
1508 other_controls = list(self.app.layout.find_all_controls())
1509 key = current_window, frozenset(other_controls)
1511 return self._cache.get(
1512 key, lambda: self._create_key_bindings(current_window, other_controls)
1513 )
1515 def get_bindings_for_keys(self, keys: KeysTuple) -> list[Binding]:
1516 return self._key_bindings.get_bindings_for_keys(keys)
1518 def get_bindings_starting_with_keys(self, keys: KeysTuple) -> list[Binding]:
1519 return self._key_bindings.get_bindings_starting_with_keys(keys)
1522async def _do_wait_for_enter(wait_text: AnyFormattedText) -> None:
1523 """
1524 Create a sub application to wait for the enter key press.
1525 This has two advantages over using 'input'/'raw_input':
1526 - This will share the same input/output I/O.
1527 - This doesn't block the event loop.
1528 """
1529 from prompt_toolkit.shortcuts import PromptSession
1531 key_bindings = KeyBindings()
1533 @key_bindings.add("enter")
1534 def _ok(event: E) -> None:
1535 event.app.exit()
1537 @key_bindings.add(Keys.Any)
1538 def _ignore(event: E) -> None:
1539 "Disallow typing."
1540 pass
1542 session: PromptSession[None] = PromptSession(
1543 message=wait_text, key_bindings=key_bindings
1544 )
1545 try:
1546 await session.app.run_async()
1547 except KeyboardInterrupt:
1548 pass # Control-c pressed. Don't propagate this error.
1551@contextmanager
1552def attach_winch_signal_handler(
1553 handler: Callable[[], None],
1554) -> Generator[None, None, None]:
1555 """
1556 Attach the given callback as a WINCH signal handler within the context
1557 manager. Restore the original signal handler when done.
1559 The `Application.run` method will register SIGWINCH, so that it will
1560 properly repaint when the terminal window resizes. However, using
1561 `run_in_terminal`, we can temporarily send an application to the
1562 background, and run an other app in between, which will then overwrite the
1563 SIGWINCH. This is why it's important to restore the handler when the app
1564 terminates.
1565 """
1566 # The tricky part here is that signals are registered in the Unix event
1567 # loop with a wakeup fd, but another application could have registered
1568 # signals using signal.signal directly. For now, the implementation is
1569 # hard-coded for the `asyncio.unix_events._UnixSelectorEventLoop`.
1571 # No WINCH? Then don't do anything.
1572 sigwinch = getattr(signal, "SIGWINCH", None)
1573 if sigwinch is None or not in_main_thread():
1574 yield
1575 return
1577 # Keep track of the previous handler.
1578 # (Only UnixSelectorEventloop has `_signal_handlers`.)
1579 loop = get_running_loop()
1580 previous_winch_handler = getattr(loop, "_signal_handlers", {}).get(sigwinch)
1582 try:
1583 loop.add_signal_handler(sigwinch, handler)
1584 yield
1585 finally:
1586 # Restore the previous signal handler.
1587 loop.remove_signal_handler(sigwinch)
1588 if previous_winch_handler is not None:
1589 loop.add_signal_handler(
1590 sigwinch,
1591 previous_winch_handler._callback,
1592 *previous_winch_handler._args,
1593 )
1596@contextmanager
1597def _restore_sigint_from_ctypes() -> Generator[None, None, None]:
1598 # The following functions are part of the stable ABI since python 3.2
1599 # See: https://docs.python.org/3/c-api/sys.html#c.PyOS_getsig
1600 # Inline import: these are not available on Pypy.
1601 try:
1602 from ctypes import c_int, c_void_p, pythonapi
1603 except ImportError:
1604 have_ctypes_signal = False
1605 else:
1606 # GraalPy has the functions, but they don't work
1607 have_ctypes_signal = sys.implementation.name != "graalpy"
1609 if have_ctypes_signal:
1610 # PyOS_sighandler_t PyOS_getsig(int i)
1611 pythonapi.PyOS_getsig.restype = c_void_p
1612 pythonapi.PyOS_getsig.argtypes = (c_int,)
1614 # PyOS_sighandler_t PyOS_setsig(int i, PyOS_sighandler_t h)
1615 pythonapi.PyOS_setsig.restype = c_void_p
1616 pythonapi.PyOS_setsig.argtypes = (
1617 c_int,
1618 c_void_p,
1619 )
1621 sigint = signal.getsignal(signal.SIGINT)
1622 if have_ctypes_signal:
1623 sigint_os = pythonapi.PyOS_getsig(signal.SIGINT)
1625 try:
1626 yield
1627 finally:
1628 if sigint is not None:
1629 signal.signal(signal.SIGINT, sigint)
1630 if have_ctypes_signal:
1631 pythonapi.PyOS_setsig(signal.SIGINT, sigint_os)