Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/prompt_toolkit/application/run_in_terminal.py: 26%
42 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-20 06:09 +0000
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-20 06:09 +0000
1"""
2Tools for running functions on the terminal above the current application or prompt.
3"""
4from __future__ import annotations
6from asyncio import Future, ensure_future
7from contextlib import asynccontextmanager
8from typing import AsyncGenerator, Awaitable, Callable, TypeVar
10from prompt_toolkit.eventloop import run_in_executor_with_context
12from .current import get_app_or_none
14__all__ = [
15 "run_in_terminal",
16 "in_terminal",
17]
19_T = TypeVar("_T")
22def run_in_terminal(
23 func: Callable[[], _T], render_cli_done: bool = False, in_executor: bool = False
24) -> Awaitable[_T]:
25 """
26 Run function on the terminal above the current application or prompt.
28 What this does is first hiding the prompt, then running this callable
29 (which can safely output to the terminal), and then again rendering the
30 prompt which causes the output of this function to scroll above the
31 prompt.
33 ``func`` is supposed to be a synchronous function. If you need an
34 asynchronous version of this function, use the ``in_terminal`` context
35 manager directly.
37 :param func: The callable to execute.
38 :param render_cli_done: When True, render the interface in the
39 'Done' state first, then execute the function. If False,
40 erase the interface first.
41 :param in_executor: When True, run in executor. (Use this for long
42 blocking functions, when you don't want to block the event loop.)
44 :returns: A `Future`.
45 """
47 async def run() -> _T:
48 async with in_terminal(render_cli_done=render_cli_done):
49 if in_executor:
50 return await run_in_executor_with_context(func)
51 else:
52 return func()
54 return ensure_future(run())
57@asynccontextmanager
58async def in_terminal(render_cli_done: bool = False) -> AsyncGenerator[None, None]:
59 """
60 Asynchronous context manager that suspends the current application and runs
61 the body in the terminal.
63 .. code::
65 async def f():
66 async with in_terminal():
67 call_some_function()
68 await call_some_async_function()
69 """
70 app = get_app_or_none()
71 if app is None or not app._is_running:
72 yield
73 return
75 # When a previous `run_in_terminal` call was in progress. Wait for that
76 # to finish, before starting this one. Chain to previous call.
77 previous_run_in_terminal_f = app._running_in_terminal_f
78 new_run_in_terminal_f: Future[None] = Future()
79 app._running_in_terminal_f = new_run_in_terminal_f
81 # Wait for the previous `run_in_terminal` to finish.
82 if previous_run_in_terminal_f is not None:
83 await previous_run_in_terminal_f
85 # Wait for all CPRs to arrive. We don't want to detach the input until
86 # all cursor position responses have been arrived. Otherwise, the tty
87 # will echo its input and can show stuff like ^[[39;1R.
88 if app.output.responds_to_cpr:
89 await app.renderer.wait_for_cpr_responses()
91 # Draw interface in 'done' state, or erase.
92 if render_cli_done:
93 app._redraw(render_as_done=True)
94 else:
95 app.renderer.erase()
97 # Disable rendering.
98 app._running_in_terminal = True
100 # Detach input.
101 try:
102 with app.input.detach():
103 with app.input.cooked_mode():
104 yield
105 finally:
106 # Redraw interface again.
107 try:
108 app._running_in_terminal = False
109 app.renderer.reset()
110 app._request_absolute_cursor_position()
111 app._redraw()
112 finally:
113 new_run_in_terminal_f.set_result(None)