summaryrefslogtreecommitdiffstats
path: root/src/prompt_toolkit/application
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 17:35:20 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-04 17:35:20 +0000
commite106bf94eff07d9a59771d9ccc4406421e18ab64 (patch)
treeedb6545500e39df9c67aa918a6125bffc8ec1aee /src/prompt_toolkit/application
parentInitial commit. (diff)
downloadprompt-toolkit-e106bf94eff07d9a59771d9ccc4406421e18ab64.tar.xz
prompt-toolkit-e106bf94eff07d9a59771d9ccc4406421e18ab64.zip
Adding upstream version 3.0.36.upstream/3.0.36upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/prompt_toolkit/application')
-rw-r--r--src/prompt_toolkit/application/__init__.py30
-rw-r--r--src/prompt_toolkit/application/application.py1521
-rw-r--r--src/prompt_toolkit/application/current.py197
-rw-r--r--src/prompt_toolkit/application/dummy.py51
-rw-r--r--src/prompt_toolkit/application/run_in_terminal.py117
5 files changed, 1916 insertions, 0 deletions
diff --git a/src/prompt_toolkit/application/__init__.py b/src/prompt_toolkit/application/__init__.py
new file mode 100644
index 0000000..dc61ca7
--- /dev/null
+++ b/src/prompt_toolkit/application/__init__.py
@@ -0,0 +1,30 @@
+from .application import Application
+from .current import (
+ AppSession,
+ create_app_session,
+ create_app_session_from_tty,
+ get_app,
+ get_app_or_none,
+ get_app_session,
+ set_app,
+)
+from .dummy import DummyApplication
+from .run_in_terminal import in_terminal, run_in_terminal
+
+__all__ = [
+ # Application.
+ "Application",
+ # Current.
+ "AppSession",
+ "get_app_session",
+ "create_app_session",
+ "create_app_session_from_tty",
+ "get_app",
+ "get_app_or_none",
+ "set_app",
+ # Dummy.
+ "DummyApplication",
+ # Run_in_terminal
+ "in_terminal",
+ "run_in_terminal",
+]
diff --git a/src/prompt_toolkit/application/application.py b/src/prompt_toolkit/application/application.py
new file mode 100644
index 0000000..0915a9e
--- /dev/null
+++ b/src/prompt_toolkit/application/application.py
@@ -0,0 +1,1521 @@
+import asyncio
+import os
+import re
+import signal
+import sys
+import threading
+import time
+from asyncio import (
+ AbstractEventLoop,
+ Future,
+ Task,
+ ensure_future,
+ new_event_loop,
+ set_event_loop,
+ sleep,
+)
+from contextlib import ExitStack, contextmanager
+from subprocess import Popen
+from traceback import format_tb
+from typing import (
+ Any,
+ Awaitable,
+ Callable,
+ Coroutine,
+ Dict,
+ FrozenSet,
+ Generator,
+ Generic,
+ Hashable,
+ Iterable,
+ Iterator,
+ List,
+ Optional,
+ Set,
+ Tuple,
+ Type,
+ TypeVar,
+ Union,
+ cast,
+ overload,
+)
+
+from prompt_toolkit.buffer import Buffer
+from prompt_toolkit.cache import SimpleCache
+from prompt_toolkit.clipboard import Clipboard, InMemoryClipboard
+from prompt_toolkit.cursor_shapes import AnyCursorShapeConfig, to_cursor_shape_config
+from prompt_toolkit.data_structures import Size
+from prompt_toolkit.enums import EditingMode
+from prompt_toolkit.eventloop import (
+ get_traceback_from_context,
+ run_in_executor_with_context,
+)
+from prompt_toolkit.eventloop.utils import call_soon_threadsafe, get_event_loop
+from prompt_toolkit.filters import Condition, Filter, FilterOrBool, to_filter
+from prompt_toolkit.formatted_text import AnyFormattedText
+from prompt_toolkit.input.base import Input
+from prompt_toolkit.input.typeahead import get_typeahead, store_typeahead
+from prompt_toolkit.key_binding.bindings.page_navigation import (
+ load_page_navigation_bindings,
+)
+from prompt_toolkit.key_binding.defaults import load_key_bindings
+from prompt_toolkit.key_binding.emacs_state import EmacsState
+from prompt_toolkit.key_binding.key_bindings import (
+ Binding,
+ ConditionalKeyBindings,
+ GlobalOnlyKeyBindings,
+ KeyBindings,
+ KeyBindingsBase,
+ KeysTuple,
+ merge_key_bindings,
+)
+from prompt_toolkit.key_binding.key_processor import KeyPressEvent, KeyProcessor
+from prompt_toolkit.key_binding.vi_state import ViState
+from prompt_toolkit.keys import Keys
+from prompt_toolkit.layout.containers import Container, Window
+from prompt_toolkit.layout.controls import BufferControl, UIControl
+from prompt_toolkit.layout.dummy import create_dummy_layout
+from prompt_toolkit.layout.layout import Layout, walk
+from prompt_toolkit.output import ColorDepth, Output
+from prompt_toolkit.renderer import Renderer, print_formatted_text
+from prompt_toolkit.search import SearchState
+from prompt_toolkit.styles import (
+ BaseStyle,
+ DummyStyle,
+ DummyStyleTransformation,
+ DynamicStyle,
+ StyleTransformation,
+ default_pygments_style,
+ default_ui_style,
+ merge_styles,
+)
+from prompt_toolkit.utils import Event, in_main_thread
+
+from .current import get_app_session, set_app
+from .run_in_terminal import in_terminal, run_in_terminal
+
+try:
+ import contextvars
+except ImportError:
+ import prompt_toolkit.eventloop.dummy_contextvars as contextvars # type: ignore
+
+
+__all__ = [
+ "Application",
+]
+
+
+E = KeyPressEvent
+_AppResult = TypeVar("_AppResult")
+ApplicationEventHandler = Callable[["Application[_AppResult]"], None]
+
+_SIGWINCH = getattr(signal, "SIGWINCH", None)
+_SIGTSTP = getattr(signal, "SIGTSTP", None)
+
+
+class Application(Generic[_AppResult]):
+ """
+ The main Application class!
+ This glues everything together.
+
+ :param layout: A :class:`~prompt_toolkit.layout.Layout` instance.
+ :param key_bindings:
+ :class:`~prompt_toolkit.key_binding.KeyBindingsBase` instance for
+ the key bindings.
+ :param clipboard: :class:`~prompt_toolkit.clipboard.Clipboard` to use.
+ :param full_screen: When True, run the application on the alternate screen buffer.
+ :param color_depth: Any :class:`~.ColorDepth` value, a callable that
+ returns a :class:`~.ColorDepth` or `None` for default.
+ :param erase_when_done: (bool) Clear the application output when it finishes.
+ :param reverse_vi_search_direction: Normally, in Vi mode, a '/' searches
+ forward and a '?' searches backward. In Readline mode, this is usually
+ reversed.
+ :param min_redraw_interval: Number of seconds to wait between redraws. Use
+ this for applications where `invalidate` is called a lot. This could cause
+ a lot of terminal output, which some terminals are not able to process.
+
+ `None` means that every `invalidate` will be scheduled right away
+ (which is usually fine).
+
+ When one `invalidate` is called, but a scheduled redraw of a previous
+ `invalidate` call has not been executed yet, nothing will happen in any
+ case.
+
+ :param max_render_postpone_time: When there is high CPU (a lot of other
+ scheduled calls), postpone the rendering max x seconds. '0' means:
+ don't postpone. '.5' means: try to draw at least twice a second.
+
+ :param refresh_interval: Automatically invalidate the UI every so many
+ seconds. When `None` (the default), only invalidate when `invalidate`
+ has been called.
+
+ :param terminal_size_polling_interval: Poll the terminal size every so many
+ seconds. Useful if the applications runs in a thread other then then
+ main thread where SIGWINCH can't be handled, or on Windows.
+
+ Filters:
+
+ :param mouse_support: (:class:`~prompt_toolkit.filters.Filter` or
+ boolean). When True, enable mouse support.
+ :param paste_mode: :class:`~prompt_toolkit.filters.Filter` or boolean.
+ :param editing_mode: :class:`~prompt_toolkit.enums.EditingMode`.
+
+ :param enable_page_navigation_bindings: When `True`, enable the page
+ navigation key bindings. These include both Emacs and Vi bindings like
+ page-up, page-down and so on to scroll through pages. Mostly useful for
+ creating an editor or other full screen applications. Probably, you
+ don't want this for the implementation of a REPL. By default, this is
+ enabled if `full_screen` is set.
+
+ Callbacks (all of these should accept an
+ :class:`~prompt_toolkit.application.Application` object as input.)
+
+ :param on_reset: Called during reset.
+ :param on_invalidate: Called when the UI has been invalidated.
+ :param before_render: Called right before rendering.
+ :param after_render: Called right after rendering.
+
+ I/O:
+ (Note that the preferred way to change the input/output is by creating an
+ `AppSession` with the required input/output objects. If you need multiple
+ applications running at the same time, you have to create a separate
+ `AppSession` using a `with create_app_session():` block.
+
+ :param input: :class:`~prompt_toolkit.input.Input` instance.
+ :param output: :class:`~prompt_toolkit.output.Output` instance. (Probably
+ Vt100_Output or Win32Output.)
+
+ Usage:
+
+ app = Application(...)
+ app.run()
+
+ # Or
+ await app.run_async()
+ """
+
+ def __init__(
+ self,
+ layout: Optional[Layout] = None,
+ style: Optional[BaseStyle] = None,
+ include_default_pygments_style: FilterOrBool = True,
+ style_transformation: Optional[StyleTransformation] = None,
+ key_bindings: Optional[KeyBindingsBase] = None,
+ clipboard: Optional[Clipboard] = None,
+ full_screen: bool = False,
+ color_depth: Union[
+ ColorDepth, Callable[[], Union[ColorDepth, None]], None
+ ] = None,
+ mouse_support: FilterOrBool = False,
+ enable_page_navigation_bindings: Optional[
+ FilterOrBool
+ ] = None, # Can be None, True or False.
+ paste_mode: FilterOrBool = False,
+ editing_mode: EditingMode = EditingMode.EMACS,
+ erase_when_done: bool = False,
+ reverse_vi_search_direction: FilterOrBool = False,
+ min_redraw_interval: Union[float, int, None] = None,
+ max_render_postpone_time: Union[float, int, None] = 0.01,
+ refresh_interval: Optional[float] = None,
+ terminal_size_polling_interval: Optional[float] = 0.5,
+ cursor: AnyCursorShapeConfig = None,
+ on_reset: Optional["ApplicationEventHandler[_AppResult]"] = None,
+ on_invalidate: Optional["ApplicationEventHandler[_AppResult]"] = None,
+ before_render: Optional["ApplicationEventHandler[_AppResult]"] = None,
+ after_render: Optional["ApplicationEventHandler[_AppResult]"] = None,
+ # I/O.
+ input: Optional[Input] = None,
+ output: Optional[Output] = None,
+ ) -> None:
+
+ # If `enable_page_navigation_bindings` is not specified, enable it in
+ # case of full screen applications only. This can be overridden by the user.
+ if enable_page_navigation_bindings is None:
+ enable_page_navigation_bindings = Condition(lambda: self.full_screen)
+
+ paste_mode = to_filter(paste_mode)
+ mouse_support = to_filter(mouse_support)
+ reverse_vi_search_direction = to_filter(reverse_vi_search_direction)
+ enable_page_navigation_bindings = to_filter(enable_page_navigation_bindings)
+ include_default_pygments_style = to_filter(include_default_pygments_style)
+
+ if layout is None:
+ layout = create_dummy_layout()
+
+ if style_transformation is None:
+ style_transformation = DummyStyleTransformation()
+
+ self.style = style
+ self.style_transformation = style_transformation
+
+ # Key bindings.
+ self.key_bindings = key_bindings
+ self._default_bindings = load_key_bindings()
+ self._page_navigation_bindings = load_page_navigation_bindings()
+
+ self.layout = layout
+ self.clipboard = clipboard or InMemoryClipboard()
+ self.full_screen: bool = full_screen
+ self._color_depth = color_depth
+ self.mouse_support = mouse_support
+
+ self.paste_mode = paste_mode
+ self.editing_mode = editing_mode
+ self.erase_when_done = erase_when_done
+ self.reverse_vi_search_direction = reverse_vi_search_direction
+ self.enable_page_navigation_bindings = enable_page_navigation_bindings
+ self.min_redraw_interval = min_redraw_interval
+ self.max_render_postpone_time = max_render_postpone_time
+ self.refresh_interval = refresh_interval
+ self.terminal_size_polling_interval = terminal_size_polling_interval
+
+ self.cursor = to_cursor_shape_config(cursor)
+
+ # Events.
+ self.on_invalidate = Event(self, on_invalidate)
+ self.on_reset = Event(self, on_reset)
+ self.before_render = Event(self, before_render)
+ self.after_render = Event(self, after_render)
+
+ # I/O.
+ session = get_app_session()
+ self.output = output or session.output
+ self.input = input or session.input
+
+ # List of 'extra' functions to execute before a Application.run.
+ self.pre_run_callables: List[Callable[[], None]] = []
+
+ self._is_running = False
+ self.future: Optional[Future[_AppResult]] = None
+ self.loop: Optional[AbstractEventLoop] = None
+ self.context: Optional[contextvars.Context] = None
+
+ #: Quoted insert. This flag is set if we go into quoted insert mode.
+ self.quoted_insert = False
+
+ #: Vi state. (For Vi key bindings.)
+ self.vi_state = ViState()
+ self.emacs_state = EmacsState()
+
+ #: When to flush the input (For flushing escape keys.) This is important
+ #: on terminals that use vt100 input. We can't distinguish the escape
+ #: key from for instance the left-arrow key, if we don't know what follows
+ #: after "\x1b". This little timer will consider "\x1b" to be escape if
+ #: nothing did follow in this time span.
+ #: This seems to work like the `ttimeoutlen` option in Vim.
+ self.ttimeoutlen = 0.5 # Seconds.
+
+ #: Like Vim's `timeoutlen` option. This can be `None` or a float. For
+ #: instance, suppose that we have a key binding AB and a second key
+ #: binding A. If the uses presses A and then waits, we don't handle
+ #: this binding yet (unless it was marked 'eager'), because we don't
+ #: know what will follow. This timeout is the maximum amount of time
+ #: that we wait until we call the handlers anyway. Pass `None` to
+ #: disable this timeout.
+ self.timeoutlen = 1.0
+
+ #: The `Renderer` instance.
+ # Make sure that the same stdout is used, when a custom renderer has been passed.
+ self._merged_style = self._create_merged_style(include_default_pygments_style)
+
+ self.renderer = Renderer(
+ self._merged_style,
+ self.output,
+ full_screen=full_screen,
+ mouse_support=mouse_support,
+ cpr_not_supported_callback=self.cpr_not_supported_callback,
+ )
+
+ #: Render counter. This one is increased every time the UI is rendered.
+ #: It can be used as a key for caching certain information during one
+ #: rendering.
+ self.render_counter = 0
+
+ # Invalidate flag. When 'True', a repaint has been scheduled.
+ self._invalidated = False
+ self._invalidate_events: List[
+ Event[object]
+ ] = [] # Collection of 'invalidate' Event objects.
+ self._last_redraw_time = 0.0 # Unix timestamp of last redraw. Used when
+ # `min_redraw_interval` is given.
+
+ #: The `InputProcessor` instance.
+ self.key_processor = KeyProcessor(_CombinedRegistry(self))
+
+ # If `run_in_terminal` was called. This will point to a `Future` what will be
+ # set at the point when the previous run finishes.
+ self._running_in_terminal = False
+ self._running_in_terminal_f: Optional[Future[None]] = None
+
+ # Trigger initialize callback.
+ self.reset()
+
+ def _create_merged_style(self, include_default_pygments_style: Filter) -> BaseStyle:
+ """
+ Create a `Style` object that merges the default UI style, the default
+ pygments style, and the custom user style.
+ """
+ dummy_style = DummyStyle()
+ pygments_style = default_pygments_style()
+
+ @DynamicStyle
+ def conditional_pygments_style() -> BaseStyle:
+ if include_default_pygments_style():
+ return pygments_style
+ else:
+ return dummy_style
+
+ return merge_styles(
+ [
+ default_ui_style(),
+ conditional_pygments_style,
+ DynamicStyle(lambda: self.style),
+ ]
+ )
+
+ @property
+ def color_depth(self) -> ColorDepth:
+ """
+ The active :class:`.ColorDepth`.
+
+ The current value is determined as follows:
+
+ - If a color depth was given explicitly to this application, use that
+ value.
+ - Otherwise, fall back to the color depth that is reported by the
+ :class:`.Output` implementation. If the :class:`.Output` class was
+ created using `output.defaults.create_output`, then this value is
+ coming from the $PROMPT_TOOLKIT_COLOR_DEPTH environment variable.
+ """
+ depth = self._color_depth
+
+ if callable(depth):
+ depth = depth()
+
+ if depth is None:
+ depth = self.output.get_default_color_depth()
+
+ return depth
+
+ @property
+ def current_buffer(self) -> Buffer:
+ """
+ The currently focused :class:`~.Buffer`.
+
+ (This returns a dummy :class:`.Buffer` when none of the actual buffers
+ has the focus. In this case, it's really not practical to check for
+ `None` values or catch exceptions every time.)
+ """
+ return self.layout.current_buffer or Buffer(
+ name="dummy-buffer"
+ ) # Dummy buffer.
+
+ @property
+ def current_search_state(self) -> SearchState:
+ """
+ Return the current :class:`.SearchState`. (The one for the focused
+ :class:`.BufferControl`.)
+ """
+ ui_control = self.layout.current_control
+ if isinstance(ui_control, BufferControl):
+ return ui_control.search_state
+ else:
+ return SearchState() # Dummy search state. (Don't return None!)
+
+ def reset(self) -> None:
+ """
+ Reset everything, for reading the next input.
+ """
+ # Notice that we don't reset the buffers. (This happens just before
+ # returning, and when we have multiple buffers, we clearly want the
+ # content in the other buffers to remain unchanged between several
+ # calls of `run`. (And the same is true for the focus stack.)
+
+ self.exit_style = ""
+
+ self._background_tasks: Set[Task[None]] = set()
+
+ self.renderer.reset()
+ self.key_processor.reset()
+ self.layout.reset()
+ self.vi_state.reset()
+ self.emacs_state.reset()
+
+ # Trigger reset event.
+ self.on_reset.fire()
+
+ # Make sure that we have a 'focusable' widget focused.
+ # (The `Layout` class can't determine this.)
+ layout = self.layout
+
+ if not layout.current_control.is_focusable():
+ for w in layout.find_all_windows():
+ if w.content.is_focusable():
+ layout.current_window = w
+ break
+
+ def invalidate(self) -> None:
+ """
+ Thread safe way of sending a repaint trigger to the input event loop.
+ """
+ if not self._is_running:
+ # Don't schedule a redraw if we're not running.
+ # Otherwise, `get_event_loop()` in `call_soon_threadsafe` can fail.
+ # See: https://github.com/dbcli/mycli/issues/797
+ return
+
+ # `invalidate()` called if we don't have a loop yet (not running?), or
+ # after the event loop was closed.
+ if self.loop is None or self.loop.is_closed():
+ return
+
+ # Never schedule a second redraw, when a previous one has not yet been
+ # executed. (This should protect against other threads calling
+ # 'invalidate' many times, resulting in 100% CPU.)
+ if self._invalidated:
+ return
+ else:
+ self._invalidated = True
+
+ # Trigger event.
+ self.loop.call_soon_threadsafe(self.on_invalidate.fire)
+
+ def redraw() -> None:
+ self._invalidated = False
+ self._redraw()
+
+ def schedule_redraw() -> None:
+ call_soon_threadsafe(
+ redraw, max_postpone_time=self.max_render_postpone_time, loop=self.loop
+ )
+
+ if self.min_redraw_interval:
+ # When a minimum redraw interval is set, wait minimum this amount
+ # of time between redraws.
+ diff = time.time() - self._last_redraw_time
+ if diff < self.min_redraw_interval:
+
+ async def redraw_in_future() -> None:
+ await sleep(cast(float, self.min_redraw_interval) - diff)
+ schedule_redraw()
+
+ self.loop.call_soon_threadsafe(
+ lambda: self.create_background_task(redraw_in_future())
+ )
+ else:
+ schedule_redraw()
+ else:
+ schedule_redraw()
+
+ @property
+ def invalidated(self) -> bool:
+ "True when a redraw operation has been scheduled."
+ return self._invalidated
+
+ def _redraw(self, render_as_done: bool = False) -> None:
+ """
+ Render the command line again. (Not thread safe!) (From other threads,
+ or if unsure, use :meth:`.Application.invalidate`.)
+
+ :param render_as_done: make sure to put the cursor after the UI.
+ """
+
+ def run_in_context() -> None:
+ # Only draw when no sub application was started.
+ if self._is_running and not self._running_in_terminal:
+ if self.min_redraw_interval:
+ self._last_redraw_time = time.time()
+
+ # Render
+ self.render_counter += 1
+ self.before_render.fire()
+
+ if render_as_done:
+ if self.erase_when_done:
+ self.renderer.erase()
+ else:
+ # Draw in 'done' state and reset renderer.
+ self.renderer.render(self, self.layout, is_done=render_as_done)
+ else:
+ self.renderer.render(self, self.layout)
+
+ self.layout.update_parents_relations()
+
+ # Fire render event.
+ self.after_render.fire()
+
+ self._update_invalidate_events()
+
+ # NOTE: We want to make sure this Application is the active one. The
+ # invalidate function is often called from a context where this
+ # application is not the active one. (Like the
+ # `PromptSession._auto_refresh_context`).
+ # We copy the context in case the context was already active, to
+ # prevent RuntimeErrors. (The rendering is not supposed to change
+ # any context variables.)
+ if self.context is not None:
+ self.context.copy().run(run_in_context)
+
+ def _start_auto_refresh_task(self) -> None:
+ """
+ Start a while/true loop in the background for automatic invalidation of
+ the UI.
+ """
+ if self.refresh_interval is not None and self.refresh_interval != 0:
+
+ async def auto_refresh(refresh_interval: float) -> None:
+ while True:
+ await sleep(refresh_interval)
+ self.invalidate()
+
+ self.create_background_task(auto_refresh(self.refresh_interval))
+
+ def _update_invalidate_events(self) -> None:
+ """
+ Make sure to attach 'invalidate' handlers to all invalidate events in
+ the UI.
+ """
+ # Remove all the original event handlers. (Components can be removed
+ # from the UI.)
+ for ev in self._invalidate_events:
+ ev -= self._invalidate_handler
+
+ # Gather all new events.
+ # (All controls are able to invalidate themselves.)
+ def gather_events() -> Iterable[Event[object]]:
+ for c in self.layout.find_all_controls():
+ yield from c.get_invalidate_events()
+
+ self._invalidate_events = list(gather_events())
+
+ for ev in self._invalidate_events:
+ ev += self._invalidate_handler
+
+ def _invalidate_handler(self, sender: object) -> None:
+ """
+ Handler for invalidate events coming from UIControls.
+
+ (This handles the difference in signature between event handler and
+ `self.invalidate`. It also needs to be a method -not a nested
+ function-, so that we can remove it again .)
+ """
+ self.invalidate()
+
+ def _on_resize(self) -> None:
+ """
+ When the window size changes, we erase the current output and request
+ again the cursor position. When the CPR answer arrives, the output is
+ drawn again.
+ """
+ # Erase, request position (when cursor is at the start position)
+ # and redraw again. -- The order is important.
+ self.renderer.erase(leave_alternate_screen=False)
+ self._request_absolute_cursor_position()
+ self._redraw()
+
+ def _pre_run(self, pre_run: Optional[Callable[[], None]] = None) -> None:
+ """
+ Called during `run`.
+
+ `self.future` should be set to the new future at the point where this
+ is called in order to avoid data races. `pre_run` can be used to set a
+ `threading.Event` to synchronize with UI termination code, running in
+ another thread that would call `Application.exit`. (See the progress
+ bar code for an example.)
+ """
+ if pre_run:
+ pre_run()
+
+ # Process registered "pre_run_callables" and clear list.
+ for c in self.pre_run_callables:
+ c()
+ del self.pre_run_callables[:]
+
+ async def run_async(
+ self,
+ pre_run: Optional[Callable[[], None]] = None,
+ set_exception_handler: bool = True,
+ handle_sigint: bool = True,
+ slow_callback_duration: float = 0.5,
+ ) -> _AppResult:
+ """
+ Run the prompt_toolkit :class:`~prompt_toolkit.application.Application`
+ until :meth:`~prompt_toolkit.application.Application.exit` has been
+ called. Return the value that was passed to
+ :meth:`~prompt_toolkit.application.Application.exit`.
+
+ This is the main entry point for a prompt_toolkit
+ :class:`~prompt_toolkit.application.Application` and usually the only
+ place where the event loop is actually running.
+
+ :param pre_run: Optional callable, which is called right after the
+ "reset" of the application.
+ :param set_exception_handler: When set, in case of an exception, go out
+ of the alternate screen and hide the application, display the
+ exception, and wait for the user to press ENTER.
+ :param handle_sigint: Handle SIGINT signal if possible. This will call
+ the `<sigint>` key binding when a SIGINT is received. (This only
+ works in the main thread.)
+ :param slow_callback_duration: Display warnings if code scheduled in
+ the asyncio event loop takes more time than this. The asyncio
+ default of `0.1` is sometimes not sufficient on a slow system,
+ because exceptionally, the drawing of the app, which happens in the
+ event loop, can take a bit longer from time to time.
+ """
+ assert not self._is_running, "Application is already running."
+
+ if not in_main_thread() or sys.platform == "win32":
+ # Handling signals in other threads is not supported.
+ # Also on Windows, `add_signal_handler(signal.SIGINT, ...)` raises
+ # `NotImplementedError`.
+ # See: https://github.com/prompt-toolkit/python-prompt-toolkit/issues/1553
+ handle_sigint = False
+
+ async def _run_async(f: "asyncio.Future[_AppResult]") -> _AppResult:
+ self.context = contextvars.copy_context()
+
+ # Counter for cancelling 'flush' timeouts. Every time when a key is
+ # pressed, we start a 'flush' timer for flushing our escape key. But
+ # when any subsequent input is received, a new timer is started and
+ # the current timer will be ignored.
+ flush_task: Optional[asyncio.Task[None]] = None
+
+ # Reset.
+ # (`self.future` needs to be set when `pre_run` is called.)
+ self.reset()
+ self._pre_run(pre_run)
+
+ # Feed type ahead input first.
+ self.key_processor.feed_multiple(get_typeahead(self.input))
+ self.key_processor.process_keys()
+
+ def read_from_input() -> None:
+ nonlocal flush_task
+
+ # Ignore when we aren't running anymore. This callback will
+ # removed from the loop next time. (It could be that it was
+ # still in the 'tasks' list of the loop.)
+ # Except: if we need to process incoming CPRs.
+ if not self._is_running and not self.renderer.waiting_for_cpr:
+ return
+
+ # Get keys from the input object.
+ keys = self.input.read_keys()
+
+ # Feed to key processor.
+ self.key_processor.feed_multiple(keys)
+ self.key_processor.process_keys()
+
+ # Quit when the input stream was closed.
+ if self.input.closed:
+ if not f.done():
+ f.set_exception(EOFError)
+ else:
+ # Automatically flush keys.
+ if flush_task:
+ flush_task.cancel()
+ flush_task = self.create_background_task(auto_flush_input())
+
+ async def auto_flush_input() -> None:
+ # Flush input after timeout.
+ # (Used for flushing the enter key.)
+ # This sleep can be cancelled, in that case we won't flush yet.
+ await sleep(self.ttimeoutlen)
+ flush_input()
+
+ def flush_input() -> None:
+ if not self.is_done:
+ # Get keys, and feed to key processor.
+ keys = self.input.flush_keys()
+ self.key_processor.feed_multiple(keys)
+ self.key_processor.process_keys()
+
+ if self.input.closed:
+ f.set_exception(EOFError)
+
+ # Enter raw mode, attach input and attach WINCH event handler.
+ with self.input.raw_mode(), self.input.attach(
+ read_from_input
+ ), attach_winch_signal_handler(self._on_resize):
+
+ # Draw UI.
+ self._request_absolute_cursor_position()
+ self._redraw()
+ self._start_auto_refresh_task()
+
+ self.create_background_task(self._poll_output_size())
+
+ # Wait for UI to finish.
+ try:
+ result = await f
+ finally:
+ # In any case, when the application finishes.
+ # (Successful, or because of an error.)
+ try:
+ self._redraw(render_as_done=True)
+ finally:
+ # _redraw has a good chance to fail if it calls widgets
+ # with bad code. Make sure to reset the renderer
+ # anyway.
+ self.renderer.reset()
+
+ # Unset `is_running`, this ensures that possibly
+ # scheduled draws won't paint during the following
+ # yield.
+ self._is_running = False
+
+ # Detach event handlers for invalidate events.
+ # (Important when a UIControl is embedded in multiple
+ # applications, like ptterm in pymux. An invalidate
+ # should not trigger a repaint in terminated
+ # applications.)
+ for ev in self._invalidate_events:
+ ev -= self._invalidate_handler
+ self._invalidate_events = []
+
+ # Wait for CPR responses.
+ if self.output.responds_to_cpr:
+ await self.renderer.wait_for_cpr_responses()
+
+ # Wait for the run-in-terminals to terminate.
+ previous_run_in_terminal_f = self._running_in_terminal_f
+
+ if previous_run_in_terminal_f:
+ await previous_run_in_terminal_f
+
+ # Store unprocessed input as typeahead for next time.
+ store_typeahead(self.input, self.key_processor.empty_queue())
+
+ return result
+
+ @contextmanager
+ def get_loop() -> Iterator[AbstractEventLoop]:
+ loop = get_event_loop()
+ self.loop = loop
+
+ try:
+ yield loop
+ finally:
+ self.loop = None
+
+ @contextmanager
+ def set_is_running() -> Iterator[None]:
+ self._is_running = True
+ try:
+ yield
+ finally:
+ self._is_running = False
+
+ @contextmanager
+ def set_handle_sigint(loop: AbstractEventLoop) -> Iterator[None]:
+ if handle_sigint:
+ loop.add_signal_handler(
+ signal.SIGINT,
+ lambda *_: loop.call_soon_threadsafe(
+ self.key_processor.send_sigint
+ ),
+ )
+ try:
+ yield
+ finally:
+ loop.remove_signal_handler(signal.SIGINT)
+ else:
+ yield
+
+ @contextmanager
+ def set_exception_handler_ctx(loop: AbstractEventLoop) -> Iterator[None]:
+ if set_exception_handler:
+ previous_exc_handler = loop.get_exception_handler()
+ loop.set_exception_handler(self._handle_exception)
+ try:
+ yield
+ finally:
+ loop.set_exception_handler(previous_exc_handler)
+
+ else:
+ yield
+
+ @contextmanager
+ def set_callback_duration(loop: AbstractEventLoop) -> Iterator[None]:
+ # Set slow_callback_duration.
+ original_slow_callback_duration = loop.slow_callback_duration
+ loop.slow_callback_duration = slow_callback_duration
+ try:
+ yield
+ finally:
+ # Reset slow_callback_duration.
+ loop.slow_callback_duration = original_slow_callback_duration
+
+ @contextmanager
+ def create_future(
+ loop: AbstractEventLoop,
+ ) -> "Iterator[asyncio.Future[_AppResult]]":
+ f = loop.create_future()
+ self.future = f # XXX: make sure to set this before calling '_redraw'.
+
+ try:
+ yield f
+ finally:
+ # Also remove the Future again. (This brings the
+ # application back to its initial state, where it also
+ # doesn't have a Future.)
+ self.future = None
+
+ with ExitStack() as stack:
+ stack.enter_context(set_is_running())
+
+ # Make sure to set `_invalidated` to `False` to begin with,
+ # otherwise we're not going to paint anything. This can happen if
+ # this application had run before on a different event loop, and a
+ # paint was scheduled using `call_soon_threadsafe` with
+ # `max_postpone_time`.
+ self._invalidated = False
+
+ loop = stack.enter_context(get_loop())
+
+ stack.enter_context(set_handle_sigint(loop))
+ stack.enter_context(set_exception_handler_ctx(loop))
+ stack.enter_context(set_callback_duration(loop))
+ stack.enter_context(set_app(self))
+ stack.enter_context(self._enable_breakpointhook())
+
+ f = stack.enter_context(create_future(loop))
+
+ try:
+ return await _run_async(f)
+ finally:
+ # Wait for the background tasks to be done. This needs to
+ # go in the finally! If `_run_async` raises
+ # `KeyboardInterrupt`, we still want to wait for the
+ # background tasks.
+ await self.cancel_and_wait_for_background_tasks()
+
+ # The `ExitStack` above is defined in typeshed in a way that it can
+ # swallow exceptions. Without next line, mypy would think that there's
+ # a possibility we don't return here. See:
+ # https://github.com/python/mypy/issues/7726
+ assert False, "unreachable"
+
+ def run(
+ self,
+ pre_run: Optional[Callable[[], None]] = None,
+ set_exception_handler: bool = True,
+ handle_sigint: bool = True,
+ in_thread: bool = False,
+ ) -> _AppResult:
+ """
+ A blocking 'run' call that waits until the UI is finished.
+
+ This will start the current asyncio event loop. If no loop is set for
+ the current thread, then it will create a new loop. If a new loop was
+ created, this won't close the new loop (if `in_thread=False`).
+
+ :param pre_run: Optional callable, which is called right after the
+ "reset" of the application.
+ :param set_exception_handler: When set, in case of an exception, go out
+ of the alternate screen and hide the application, display the
+ exception, and wait for the user to press ENTER.
+ :param in_thread: When true, run the application in a background
+ thread, and block the current thread until the application
+ terminates. This is useful if we need to be sure the application
+ won't use the current event loop (asyncio does not support nested
+ event loops). A new event loop will be created in this background
+ thread, and that loop will also be closed when the background
+ thread terminates. When this is used, it's especially important to
+ make sure that all asyncio background tasks are managed through
+ `get_appp().create_background_task()`, so that unfinished tasks are
+ properly cancelled before the event loop is closed. This is used
+ for instance in ptpython.
+ :param handle_sigint: Handle SIGINT signal. Call the key binding for
+ `Keys.SIGINT`. (This only works in the main thread.)
+ """
+ if in_thread:
+ result: _AppResult
+ exception: Optional[BaseException] = None
+
+ def run_in_thread() -> None:
+ nonlocal result, exception
+ try:
+ result = self.run(
+ pre_run=pre_run,
+ set_exception_handler=set_exception_handler,
+ # Signal handling only works in the main thread.
+ handle_sigint=False,
+ )
+ except BaseException as e:
+ exception = e
+ finally:
+ # Make sure to close the event loop in this thread. Running
+ # the application creates a new loop (because we're in
+ # another thread), but it doesn't get closed automatically
+ # (also not by the garbage collector).
+ loop = get_event_loop()
+ loop.run_until_complete(loop.shutdown_asyncgens())
+ loop.close()
+
+ thread = threading.Thread(target=run_in_thread)
+ thread.start()
+ thread.join()
+
+ if exception is not None:
+ raise exception
+ return result
+
+ # We don't create a new event loop by default, because we want to be
+ # sure that when this is called multiple times, each call of `run()`
+ # goes through the same event loop. This way, users can schedule
+ # background-tasks that keep running across multiple prompts.
+ try:
+ loop = get_event_loop()
+ except RuntimeError:
+ # Possibly we are not running in the main thread, where no event
+ # loop is set by default. Or somebody called `asyncio.run()`
+ # before, which closes the existing event loop. We can create a new
+ # loop.
+ loop = new_event_loop()
+ set_event_loop(loop)
+
+ return loop.run_until_complete(
+ self.run_async(
+ pre_run=pre_run,
+ set_exception_handler=set_exception_handler,
+ handle_sigint=handle_sigint,
+ )
+ )
+
+ def _handle_exception(
+ self, loop: AbstractEventLoop, context: Dict[str, Any]
+ ) -> None:
+ """
+ Handler for event loop exceptions.
+ This will print the exception, using run_in_terminal.
+ """
+ # For Python 2: we have to get traceback at this point, because
+ # we're still in the 'except:' block of the event loop where the
+ # traceback is still available. Moving this code in the
+ # 'print_exception' coroutine will loose the exception.
+ tb = get_traceback_from_context(context)
+ formatted_tb = "".join(format_tb(tb))
+
+ async def in_term() -> None:
+ async with in_terminal():
+ # Print output. Similar to 'loop.default_exception_handler',
+ # but don't use logger. (This works better on Python 2.)
+ print("\nUnhandled exception in event loop:")
+ print(formatted_tb)
+ print("Exception {}".format(context.get("exception")))
+
+ await _do_wait_for_enter("Press ENTER to continue...")
+
+ ensure_future(in_term())
+
+ @contextmanager
+ def _enable_breakpointhook(self) -> Generator[None, None, None]:
+ """
+ Install our custom breakpointhook for the duration of this context
+ manager. (We will only install the hook if no other custom hook was
+ set.)
+ """
+ if sys.version_info >= (3, 7) and sys.breakpointhook == sys.__breakpointhook__:
+ sys.breakpointhook = self._breakpointhook
+
+ try:
+ yield
+ finally:
+ sys.breakpointhook = sys.__breakpointhook__
+ else:
+ yield
+
+ def _breakpointhook(self, *a: object, **kw: object) -> None:
+ """
+ Breakpointhook which uses PDB, but ensures that the application is
+ hidden and input echoing is restored during each debugger dispatch.
+ """
+ app = self
+ # Inline import on purpose. We don't want to import pdb, if not needed.
+ import pdb
+ from types import FrameType
+
+ TraceDispatch = Callable[[FrameType, str, Any], Any]
+
+ class CustomPdb(pdb.Pdb):
+ def trace_dispatch(
+ self, frame: FrameType, event: str, arg: Any
+ ) -> TraceDispatch:
+ # Hide application.
+ app.renderer.erase()
+
+ # Detach input and dispatch to debugger.
+ with app.input.detach():
+ with app.input.cooked_mode():
+ return super().trace_dispatch(frame, event, arg)
+
+ # Note: we don't render the application again here, because
+ # there's a good chance that there's a breakpoint on the next
+ # line. This paint/erase cycle would move the PDB prompt back
+ # to the middle of the screen.
+
+ frame = sys._getframe().f_back
+ CustomPdb(stdout=sys.__stdout__).set_trace(frame)
+
+ def create_background_task(
+ self, coroutine: Coroutine[Any, Any, None]
+ ) -> "asyncio.Task[None]":
+ """
+ Start a background task (coroutine) for the running application. When
+ the `Application` terminates, unfinished background tasks will be
+ cancelled.
+
+ Given that we still support Python versions before 3.11, we can't use
+ task groups (and exception groups), because of that, these background
+ tasks are not allowed to raise exceptions. If they do, we'll call the
+ default exception handler from the event loop.
+
+ If at some point, we have Python 3.11 as the minimum supported Python
+ version, then we can use a `TaskGroup` (with the lifetime of
+ `Application.run_async()`, and run run the background tasks in there.
+
+ This is not threadsafe.
+ """
+ loop = self.loop or get_event_loop()
+ task: asyncio.Task[None] = loop.create_task(coroutine)
+ self._background_tasks.add(task)
+
+ task.add_done_callback(self._on_background_task_done)
+ return task
+
+ def _on_background_task_done(self, task: "asyncio.Task[None]") -> None:
+ """
+ Called when a background task completes. Remove it from
+ `_background_tasks`, and handle exceptions if any.
+ """
+ self._background_tasks.discard(task)
+
+ if task.cancelled():
+ return
+
+ exc = task.exception()
+ if exc is not None:
+ get_event_loop().call_exception_handler(
+ {
+ "message": f"prompt_toolkit.Application background task {task!r} "
+ "raised an unexpected exception.",
+ "exception": exc,
+ "task": task,
+ }
+ )
+
+ async def cancel_and_wait_for_background_tasks(self) -> None:
+ """
+ Cancel all background tasks, and wait for the cancellation to complete.
+ If any of the background tasks raised an exception, this will also
+ propagate the exception.
+
+ (If we had nurseries like Trio, this would be the `__aexit__` of a
+ nursery.)
+ """
+ for task in self._background_tasks:
+ task.cancel()
+
+ # Wait until the cancellation of the background tasks completes.
+ # `asyncio.wait()` does not propagate exceptions raised within any of
+ # these tasks, which is what we want. Otherwise, we can't distinguish
+ # between a `CancelledError` raised in this task because it got
+ # cancelled, and a `CancelledError` raised on this `await` checkpoint,
+ # because *we* got cancelled during the teardown of the application.
+ # (If we get cancelled here, then it's important to not suppress the
+ # `CancelledError`, and have it propagate.)
+ # NOTE: Currently, if we get cancelled at this point then we can't wait
+ # for the cancellation to complete (in the future, we should be
+ # using anyio or Python's 3.11 TaskGroup.)
+ # Also, if we had exception groups, we could propagate an
+ # `ExceptionGroup` if something went wrong here. Right now, we
+ # don't propagate exceptions, but have them printed in
+ # `_on_background_task_done`.
+ if len(self._background_tasks) > 0:
+ await asyncio.wait(
+ self._background_tasks, timeout=None, return_when=asyncio.ALL_COMPLETED
+ )
+
+ async def _poll_output_size(self) -> None:
+ """
+ Coroutine for polling the terminal dimensions.
+
+ Useful for situations where `attach_winch_signal_handler` is not sufficient:
+ - If we are not running in the main thread.
+ - On Windows.
+ """
+ size: Optional[Size] = None
+ interval = self.terminal_size_polling_interval
+
+ if interval is None:
+ return
+
+ while True:
+ await asyncio.sleep(interval)
+ new_size = self.output.get_size()
+
+ if size is not None and new_size != size:
+ self._on_resize()
+ size = new_size
+
+ def cpr_not_supported_callback(self) -> None:
+ """
+ Called when we don't receive the cursor position response in time.
+ """
+ if not self.output.responds_to_cpr:
+ return # We know about this already.
+
+ def in_terminal() -> None:
+ self.output.write(
+ "WARNING: your terminal doesn't support cursor position requests (CPR).\r\n"
+ )
+ self.output.flush()
+
+ run_in_terminal(in_terminal)
+
+ @overload
+ def exit(self) -> None:
+ "Exit without arguments."
+
+ @overload
+ def exit(self, *, result: _AppResult, style: str = "") -> None:
+ "Exit with `_AppResult`."
+
+ @overload
+ def exit(
+ self, *, exception: Union[BaseException, Type[BaseException]], style: str = ""
+ ) -> None:
+ "Exit with exception."
+
+ def exit(
+ self,
+ result: Optional[_AppResult] = None,
+ exception: Optional[Union[BaseException, Type[BaseException]]] = None,
+ style: str = "",
+ ) -> None:
+ """
+ Exit application.
+
+ .. note::
+
+ If `Application.exit` is called before `Application.run()` is
+ called, then the `Application` won't exit (because the
+ `Application.future` doesn't correspond to the current run). Use a
+ `pre_run` hook and an event to synchronize the closing if there's a
+ chance this can happen.
+
+ :param result: Set this result for the application.
+ :param exception: Set this exception as the result for an application. For
+ a prompt, this is often `EOFError` or `KeyboardInterrupt`.
+ :param style: Apply this style on the whole content when quitting,
+ often this is 'class:exiting' for a prompt. (Used when
+ `erase_when_done` is not set.)
+ """
+ assert result is None or exception is None
+
+ if self.future is None:
+ raise Exception("Application is not running. Application.exit() failed.")
+
+ if self.future.done():
+ raise Exception("Return value already set. Application.exit() failed.")
+
+ self.exit_style = style
+
+ if exception is not None:
+ self.future.set_exception(exception)
+ else:
+ self.future.set_result(cast(_AppResult, result))
+
+ def _request_absolute_cursor_position(self) -> None:
+ """
+ Send CPR request.
+ """
+ # Note: only do this if the input queue is not empty, and a return
+ # value has not been set. Otherwise, we won't be able to read the
+ # response anyway.
+ if not self.key_processor.input_queue and not self.is_done:
+ self.renderer.request_absolute_cursor_position()
+
+ async def run_system_command(
+ self,
+ command: str,
+ wait_for_enter: bool = True,
+ display_before_text: AnyFormattedText = "",
+ wait_text: str = "Press ENTER to continue...",
+ ) -> None:
+ """
+ Run system command (While hiding the prompt. When finished, all the
+ output will scroll above the prompt.)
+
+ :param command: Shell command to be executed.
+ :param wait_for_enter: FWait for the user to press enter, when the
+ command is finished.
+ :param display_before_text: If given, text to be displayed before the
+ command executes.
+ :return: A `Future` object.
+ """
+ async with in_terminal():
+ # Try to use the same input/output file descriptors as the one,
+ # used to run this application.
+ try:
+ input_fd = self.input.fileno()
+ except AttributeError:
+ input_fd = sys.stdin.fileno()
+ try:
+ output_fd = self.output.fileno()
+ except AttributeError:
+ output_fd = sys.stdout.fileno()
+
+ # Run sub process.
+ def run_command() -> None:
+ self.print_text(display_before_text)
+ p = Popen(command, shell=True, stdin=input_fd, stdout=output_fd)
+ p.wait()
+
+ await run_in_executor_with_context(run_command)
+
+ # Wait for the user to press enter.
+ if wait_for_enter:
+ await _do_wait_for_enter(wait_text)
+
+ def suspend_to_background(self, suspend_group: bool = True) -> None:
+ """
+ (Not thread safe -- to be called from inside the key bindings.)
+ Suspend process.
+
+ :param suspend_group: When true, suspend the whole process group.
+ (This is the default, and probably what you want.)
+ """
+ # Only suspend when the operating system supports it.
+ # (Not on Windows.)
+ if _SIGTSTP is not None:
+
+ def run() -> None:
+ signal = cast(int, _SIGTSTP)
+ # Send `SIGTSTP` to own process.
+ # This will cause it to suspend.
+
+ # Usually we want the whole process group to be suspended. This
+ # handles the case when input is piped from another process.
+ if suspend_group:
+ os.kill(0, signal)
+ else:
+ os.kill(os.getpid(), signal)
+
+ run_in_terminal(run)
+
+ def print_text(
+ self, text: AnyFormattedText, style: Optional[BaseStyle] = None
+ ) -> None:
+ """
+ Print a list of (style_str, text) tuples to the output.
+ (When the UI is running, this method has to be called through
+ `run_in_terminal`, otherwise it will destroy the UI.)
+
+ :param text: List of ``(style_str, text)`` tuples.
+ :param style: Style class to use. Defaults to the active style in the CLI.
+ """
+ print_formatted_text(
+ output=self.output,
+ formatted_text=text,
+ style=style or self._merged_style,
+ color_depth=self.color_depth,
+ style_transformation=self.style_transformation,
+ )
+
+ @property
+ def is_running(self) -> bool:
+ "`True` when the application is currently active/running."
+ return self._is_running
+
+ @property
+ def is_done(self) -> bool:
+ if self.future:
+ return self.future.done()
+ return False
+
+ def get_used_style_strings(self) -> List[str]:
+ """
+ Return a list of used style strings. This is helpful for debugging, and
+ for writing a new `Style`.
+ """
+ attrs_for_style = self.renderer._attrs_for_style
+
+ if attrs_for_style:
+ return sorted(
+ re.sub(r"\s+", " ", style_str).strip()
+ for style_str in attrs_for_style.keys()
+ )
+
+ return []
+
+
+class _CombinedRegistry(KeyBindingsBase):
+ """
+ The `KeyBindings` of key bindings for a `Application`.
+ This merges the global key bindings with the one of the current user
+ control.
+ """
+
+ def __init__(self, app: Application[_AppResult]) -> None:
+ self.app = app
+ self._cache: SimpleCache[
+ Tuple[Window, FrozenSet[UIControl]], KeyBindingsBase
+ ] = SimpleCache()
+
+ @property
+ def _version(self) -> Hashable:
+ """Not needed - this object is not going to be wrapped in another
+ KeyBindings object."""
+ raise NotImplementedError
+
+ @property
+ def bindings(self) -> List[Binding]:
+ """Not needed - this object is not going to be wrapped in another
+ KeyBindings object."""
+ raise NotImplementedError
+
+ def _create_key_bindings(
+ self, current_window: Window, other_controls: List[UIControl]
+ ) -> KeyBindingsBase:
+ """
+ Create a `KeyBindings` object that merges the `KeyBindings` from the
+ `UIControl` with all the parent controls and the global key bindings.
+ """
+ key_bindings = []
+ collected_containers = set()
+
+ # Collect key bindings from currently focused control and all parent
+ # controls. Don't include key bindings of container parent controls.
+ container: Container = current_window
+ while True:
+ collected_containers.add(container)
+ kb = container.get_key_bindings()
+ if kb is not None:
+ key_bindings.append(kb)
+
+ if container.is_modal():
+ break
+
+ parent = self.app.layout.get_parent(container)
+ if parent is None:
+ break
+ else:
+ container = parent
+
+ # Include global bindings (starting at the top-model container).
+ for c in walk(container):
+ if c not in collected_containers:
+ kb = c.get_key_bindings()
+ if kb is not None:
+ key_bindings.append(GlobalOnlyKeyBindings(kb))
+
+ # Add App key bindings
+ if self.app.key_bindings:
+ key_bindings.append(self.app.key_bindings)
+
+ # Add mouse bindings.
+ key_bindings.append(
+ ConditionalKeyBindings(
+ self.app._page_navigation_bindings,
+ self.app.enable_page_navigation_bindings,
+ )
+ )
+ key_bindings.append(self.app._default_bindings)
+
+ # Reverse this list. The current control's key bindings should come
+ # last. They need priority.
+ key_bindings = key_bindings[::-1]
+
+ return merge_key_bindings(key_bindings)
+
+ @property
+ def _key_bindings(self) -> KeyBindingsBase:
+ current_window = self.app.layout.current_window
+ other_controls = list(self.app.layout.find_all_controls())
+ key = current_window, frozenset(other_controls)
+
+ return self._cache.get(
+ key, lambda: self._create_key_bindings(current_window, other_controls)
+ )
+
+ def get_bindings_for_keys(self, keys: KeysTuple) -> List[Binding]:
+ return self._key_bindings.get_bindings_for_keys(keys)
+
+ def get_bindings_starting_with_keys(self, keys: KeysTuple) -> List[Binding]:
+ return self._key_bindings.get_bindings_starting_with_keys(keys)
+
+
+async def _do_wait_for_enter(wait_text: AnyFormattedText) -> None:
+ """
+ Create a sub application to wait for the enter key press.
+ This has two advantages over using 'input'/'raw_input':
+ - This will share the same input/output I/O.
+ - This doesn't block the event loop.
+ """
+ from prompt_toolkit.shortcuts import PromptSession
+
+ key_bindings = KeyBindings()
+
+ @key_bindings.add("enter")
+ def _ok(event: E) -> None:
+ event.app.exit()
+
+ @key_bindings.add(Keys.Any)
+ def _ignore(event: E) -> None:
+ "Disallow typing."
+ pass
+
+ session: PromptSession[None] = PromptSession(
+ message=wait_text, key_bindings=key_bindings
+ )
+ try:
+ await session.app.run_async()
+ except KeyboardInterrupt:
+ pass # Control-c pressed. Don't propagate this error.
+
+
+@contextmanager
+def attach_winch_signal_handler(
+ handler: Callable[[], None]
+) -> Generator[None, None, None]:
+ """
+ Attach the given callback as a WINCH signal handler within the context
+ manager. Restore the original signal handler when done.
+
+ The `Application.run` method will register SIGWINCH, so that it will
+ properly repaint when the terminal window resizes. However, using
+ `run_in_terminal`, we can temporarily send an application to the
+ background, and run an other app in between, which will then overwrite the
+ SIGWINCH. This is why it's important to restore the handler when the app
+ terminates.
+ """
+ # The tricky part here is that signals are registered in the Unix event
+ # loop with a wakeup fd, but another application could have registered
+ # signals using signal.signal directly. For now, the implementation is
+ # hard-coded for the `asyncio.unix_events._UnixSelectorEventLoop`.
+
+ # No WINCH? Then don't do anything.
+ sigwinch = getattr(signal, "SIGWINCH", None)
+ if sigwinch is None or not in_main_thread():
+ yield
+ return
+
+ # Keep track of the previous handler.
+ # (Only UnixSelectorEventloop has `_signal_handlers`.)
+ loop = get_event_loop()
+ previous_winch_handler = getattr(loop, "_signal_handlers", {}).get(sigwinch)
+
+ try:
+ loop.add_signal_handler(sigwinch, handler)
+ yield
+ finally:
+ # Restore the previous signal handler.
+ loop.remove_signal_handler(sigwinch)
+ if previous_winch_handler is not None:
+ loop.add_signal_handler(
+ sigwinch,
+ previous_winch_handler._callback,
+ *previous_winch_handler._args,
+ )
diff --git a/src/prompt_toolkit/application/current.py b/src/prompt_toolkit/application/current.py
new file mode 100644
index 0000000..ae69bfd
--- /dev/null
+++ b/src/prompt_toolkit/application/current.py
@@ -0,0 +1,197 @@
+import sys
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any, Generator, Optional
+
+try:
+ from contextvars import ContextVar
+except ImportError:
+ from prompt_toolkit.eventloop.dummy_contextvars import ContextVar # type: ignore
+
+if TYPE_CHECKING:
+ from prompt_toolkit.input.base import Input
+ from prompt_toolkit.output.base import Output
+
+ from .application import Application
+
+__all__ = [
+ "AppSession",
+ "get_app_session",
+ "get_app",
+ "get_app_or_none",
+ "set_app",
+ "create_app_session",
+ "create_app_session_from_tty",
+]
+
+
+class AppSession:
+ """
+ An AppSession is an interactive session, usually connected to one terminal.
+ Within one such session, interaction with many applications can happen, one
+ after the other.
+
+ The input/output device is not supposed to change during one session.
+
+ Warning: Always use the `create_app_session` function to create an
+ instance, so that it gets activated correctly.
+
+ :param input: Use this as a default input for all applications
+ running in this session, unless an input is passed to the `Application`
+ explicitely.
+ :param output: Use this as a default output.
+ """
+
+ def __init__(
+ self, input: Optional["Input"] = None, output: Optional["Output"] = None
+ ) -> None:
+
+ self._input = input
+ self._output = output
+
+ # The application will be set dynamically by the `set_app` context
+ # manager. This is called in the application itself.
+ self.app: Optional["Application[Any]"] = None
+
+ def __repr__(self) -> str:
+ return f"AppSession(app={self.app!r})"
+
+ @property
+ def input(self) -> "Input":
+ if self._input is None:
+ from prompt_toolkit.input.defaults import create_input
+
+ self._input = create_input()
+ return self._input
+
+ @property
+ def output(self) -> "Output":
+ if self._output is None:
+ from prompt_toolkit.output.defaults import create_output
+
+ self._output = create_output()
+ return self._output
+
+
+_current_app_session: ContextVar["AppSession"] = ContextVar(
+ "_current_app_session", default=AppSession()
+)
+
+
+def get_app_session() -> AppSession:
+ return _current_app_session.get()
+
+
+def get_app() -> "Application[Any]":
+ """
+ Get the current active (running) Application.
+ An :class:`.Application` is active during the
+ :meth:`.Application.run_async` call.
+
+ We assume that there can only be one :class:`.Application` active at the
+ same time. There is only one terminal window, with only one stdin and
+ stdout. This makes the code significantly easier than passing around the
+ :class:`.Application` everywhere.
+
+ If no :class:`.Application` is running, then return by default a
+ :class:`.DummyApplication`. For practical reasons, we prefer to not raise
+ an exception. This way, we don't have to check all over the place whether
+ an actual `Application` was returned.
+
+ (For applications like pymux where we can have more than one `Application`,
+ we'll use a work-around to handle that.)
+ """
+ session = _current_app_session.get()
+ if session.app is not None:
+ return session.app
+
+ from .dummy import DummyApplication
+
+ return DummyApplication()
+
+
+def get_app_or_none() -> Optional["Application[Any]"]:
+ """
+ Get the current active (running) Application, or return `None` if no
+ application is running.
+ """
+ session = _current_app_session.get()
+ return session.app
+
+
+@contextmanager
+def set_app(app: "Application[Any]") -> Generator[None, None, None]:
+ """
+ Context manager that sets the given :class:`.Application` active in an
+ `AppSession`.
+
+ This should only be called by the `Application` itself.
+ The application will automatically be active while its running. If you want
+ the application to be active in other threads/coroutines, where that's not
+ the case, use `contextvars.copy_context()`, or use `Application.context` to
+ run it in the appropriate context.
+ """
+ session = _current_app_session.get()
+
+ previous_app = session.app
+ session.app = app
+ try:
+ yield
+ finally:
+ session.app = previous_app
+
+
+@contextmanager
+def create_app_session(
+ input: Optional["Input"] = None, output: Optional["Output"] = None
+) -> Generator[AppSession, None, None]:
+ """
+ Create a separate AppSession.
+
+ This is useful if there can be multiple individual `AppSession`s going on.
+ Like in the case of an Telnet/SSH server. This functionality uses
+ contextvars and requires at least Python 3.7.
+ """
+ if sys.version_info <= (3, 6):
+ raise RuntimeError("Application sessions require Python 3.7.")
+
+ # If no input/output is specified, fall back to the current input/output,
+ # whatever that is.
+ if input is None:
+ input = get_app_session().input
+ if output is None:
+ output = get_app_session().output
+
+ # Create new `AppSession` and activate.
+ session = AppSession(input=input, output=output)
+
+ token = _current_app_session.set(session)
+ try:
+ yield session
+ finally:
+ _current_app_session.reset(token)
+
+
+@contextmanager
+def create_app_session_from_tty() -> Generator[AppSession, None, None]:
+ """
+ Create `AppSession` that always prefers the TTY input/output.
+
+ Even if `sys.stdin` and `sys.stdout` are connected to input/output pipes,
+ this will still use the terminal for interaction (because `sys.stderr` is
+ still connected to the terminal).
+
+ Usage::
+
+ from prompt_toolkit.shortcuts import prompt
+
+ with create_app_session_from_tty():
+ prompt('>')
+ """
+ from prompt_toolkit.input.defaults import create_input
+ from prompt_toolkit.output.defaults import create_output
+
+ input = create_input(always_prefer_tty=True)
+ output = create_output(always_prefer_tty=True)
+
+ with create_app_session(input=input, output=output) as app_session:
+ yield app_session
diff --git a/src/prompt_toolkit/application/dummy.py b/src/prompt_toolkit/application/dummy.py
new file mode 100644
index 0000000..4e5e4aa
--- /dev/null
+++ b/src/prompt_toolkit/application/dummy.py
@@ -0,0 +1,51 @@
+from typing import Callable, Optional
+
+from prompt_toolkit.formatted_text import AnyFormattedText
+from prompt_toolkit.input import DummyInput
+from prompt_toolkit.output import DummyOutput
+
+from .application import Application
+
+__all__ = [
+ "DummyApplication",
+]
+
+
+class DummyApplication(Application[None]):
+ """
+ When no :class:`.Application` is running,
+ :func:`.get_app` will run an instance of this :class:`.DummyApplication` instead.
+ """
+
+ def __init__(self) -> None:
+ super().__init__(output=DummyOutput(), input=DummyInput())
+
+ def run(
+ self,
+ pre_run: Optional[Callable[[], None]] = None,
+ set_exception_handler: bool = True,
+ handle_sigint: bool = True,
+ in_thread: bool = False,
+ ) -> None:
+ raise NotImplementedError("A DummyApplication is not supposed to run.")
+
+ async def run_async(
+ self,
+ pre_run: Optional[Callable[[], None]] = None,
+ set_exception_handler: bool = True,
+ handle_sigint: bool = True,
+ slow_callback_duration: float = 0.5,
+ ) -> None:
+ raise NotImplementedError("A DummyApplication is not supposed to run.")
+
+ async def run_system_command(
+ self,
+ command: str,
+ wait_for_enter: bool = True,
+ display_before_text: AnyFormattedText = "",
+ wait_text: str = "",
+ ) -> None:
+ raise NotImplementedError
+
+ def suspend_to_background(self, suspend_group: bool = True) -> None:
+ raise NotImplementedError
diff --git a/src/prompt_toolkit/application/run_in_terminal.py b/src/prompt_toolkit/application/run_in_terminal.py
new file mode 100644
index 0000000..d5ef8aa
--- /dev/null
+++ b/src/prompt_toolkit/application/run_in_terminal.py
@@ -0,0 +1,117 @@
+"""
+Tools for running functions on the terminal above the current application or prompt.
+"""
+import sys
+from asyncio import Future, ensure_future
+from typing import AsyncGenerator, Awaitable, Callable, TypeVar
+
+from prompt_toolkit.eventloop import run_in_executor_with_context
+
+from .current import get_app_or_none
+
+if sys.version_info >= (3, 7):
+ from contextlib import asynccontextmanager
+else:
+ from prompt_toolkit.eventloop.async_context_manager import asynccontextmanager
+
+
+__all__ = [
+ "run_in_terminal",
+ "in_terminal",
+]
+
+_T = TypeVar("_T")
+
+
+def run_in_terminal(
+ func: Callable[[], _T], render_cli_done: bool = False, in_executor: bool = False
+) -> Awaitable[_T]:
+ """
+ Run function on the terminal above the current application or prompt.
+
+ What this does is first hiding the prompt, then running this callable
+ (which can safely output to the terminal), and then again rendering the
+ prompt which causes the output of this function to scroll above the
+ prompt.
+
+ ``func`` is supposed to be a synchronous function. If you need an
+ asynchronous version of this function, use the ``in_terminal`` context
+ manager directly.
+
+ :param func: The callable to execute.
+ :param render_cli_done: When True, render the interface in the
+ 'Done' state first, then execute the function. If False,
+ erase the interface first.
+ :param in_executor: When True, run in executor. (Use this for long
+ blocking functions, when you don't want to block the event loop.)
+
+ :returns: A `Future`.
+ """
+
+ async def run() -> _T:
+ async with in_terminal(render_cli_done=render_cli_done):
+ if in_executor:
+ return await run_in_executor_with_context(func)
+ else:
+ return func()
+
+ return ensure_future(run())
+
+
+@asynccontextmanager
+async def in_terminal(render_cli_done: bool = False) -> AsyncGenerator[None, None]:
+ """
+ Asynchronous context manager that suspends the current application and runs
+ the body in the terminal.
+
+ .. code::
+
+ async def f():
+ async with in_terminal():
+ call_some_function()
+ await call_some_async_function()
+ """
+ app = get_app_or_none()
+ if app is None or not app._is_running:
+ yield
+ return
+
+ # When a previous `run_in_terminal` call was in progress. Wait for that
+ # to finish, before starting this one. Chain to previous call.
+ previous_run_in_terminal_f = app._running_in_terminal_f
+ new_run_in_terminal_f: Future[None] = Future()
+ app._running_in_terminal_f = new_run_in_terminal_f
+
+ # Wait for the previous `run_in_terminal` to finish.
+ if previous_run_in_terminal_f is not None:
+ await previous_run_in_terminal_f
+
+ # Wait for all CPRs to arrive. We don't want to detach the input until
+ # all cursor position responses have been arrived. Otherwise, the tty
+ # will echo its input and can show stuff like ^[[39;1R.
+ if app.output.responds_to_cpr:
+ await app.renderer.wait_for_cpr_responses()
+
+ # Draw interface in 'done' state, or erase.
+ if render_cli_done:
+ app._redraw(render_as_done=True)
+ else:
+ app.renderer.erase()
+
+ # Disable rendering.
+ app._running_in_terminal = True
+
+ # Detach input.
+ try:
+ with app.input.detach():
+ with app.input.cooked_mode():
+ yield
+ finally:
+ # Redraw interface again.
+ try:
+ app._running_in_terminal = False
+ app.renderer.reset()
+ app._request_absolute_cursor_position()
+ app._redraw()
+ finally:
+ new_run_in_terminal_f.set_result(None)