Adding upstream version 3.0.36.upstream/3.0.36upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

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)