diff options
Diffstat (limited to '')
-rw-r--r-- | src/prompt_toolkit/application/run_in_terminal.py | 113 |
1 files changed, 113 insertions, 0 deletions
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..1e4da2d --- /dev/null +++ b/src/prompt_toolkit/application/run_in_terminal.py @@ -0,0 +1,113 @@ +""" +Tools for running functions on the terminal above the current application or prompt. +""" +from __future__ import annotations + +from asyncio import Future, ensure_future +from contextlib import asynccontextmanager +from typing import AsyncGenerator, Awaitable, Callable, TypeVar + +from prompt_toolkit.eventloop import run_in_executor_with_context + +from .current import get_app_or_none + +__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) |