Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/prompt_toolkit/application/run_in_terminal.py: 29%

1""" 

2Tools for running functions on the terminal above the current application or prompt. 

3""" 

4 

5from __future__ import annotations 

6 

7from asyncio import Future, ensure_future 

8from collections.abc import AsyncGenerator, Awaitable, Callable 

9from contextlib import asynccontextmanager 

10from typing import TypeVar 

11 

12from prompt_toolkit.eventloop import run_in_executor_with_context 

13 

14from .current import get_app_or_none 

15 

16__all__ = [ 

17 "run_in_terminal", 

18 "in_terminal", 

19] 

20 

21_T = TypeVar("_T") 

22 

23 

24def run_in_terminal( 

25 func: Callable[[], _T], render_cli_done: bool = False, in_executor: bool = False 

26) -> Awaitable[_T]: 

27 """ 

28 Run function on the terminal above the current application or prompt. 

29 

30 What this does is first hiding the prompt, then running this callable 

31 (which can safely output to the terminal), and then again rendering the 

32 prompt which causes the output of this function to scroll above the 

33 prompt. 

34 

35 ``func`` is supposed to be a synchronous function. If you need an 

36 asynchronous version of this function, use the ``in_terminal`` context 

37 manager directly. 

38 

39 :param func: The callable to execute. 

40 :param render_cli_done: When True, render the interface in the 

41 'Done' state first, then execute the function. If False, 

42 erase the interface first. 

43 :param in_executor: When True, run in executor. (Use this for long 

44 blocking functions, when you don't want to block the event loop.) 

45 

46 :returns: A `Future`. 

47 """ 

48 

49 async def run() -> _T: 

50 async with in_terminal(render_cli_done=render_cli_done): 

51 if in_executor: 

52 return await run_in_executor_with_context(func) 

53 else: 

54 return func() 

55 

56 return ensure_future(run()) 

57 

58 

59@asynccontextmanager 

60async def in_terminal(render_cli_done: bool = False) -> AsyncGenerator[None, None]: 

61 """ 

62 Asynchronous context manager that suspends the current application and runs 

63 the body in the terminal. 

64 

65 .. code:: 

66 

67 async def f(): 

68 async with in_terminal(): 

69 call_some_function() 

70 await call_some_async_function() 

71 """ 

72 app = get_app_or_none() 

73 if app is None or not app._is_running: 

74 yield 

75 return 

76 

77 # When a previous `run_in_terminal` call was in progress. Wait for that 

78 # to finish, before starting this one. Chain to previous call. 

79 previous_run_in_terminal_f = app._running_in_terminal_f 

80 new_run_in_terminal_f: Future[None] = Future() 

81 app._running_in_terminal_f = new_run_in_terminal_f 

82 

83 # Wait for the previous `run_in_terminal` to finish. 

84 if previous_run_in_terminal_f is not None: 

85 await previous_run_in_terminal_f 

86 

87 # Wait for all CPRs to arrive. We don't want to detach the input until 

88 # all cursor position responses have been arrived. Otherwise, the tty 

89 # will echo its input and can show stuff like ^[[39;1R. 

90 if app.output.responds_to_cpr: 

91 await app.renderer.wait_for_cpr_responses() 

92 

93 # Draw interface in 'done' state, or erase. 

94 if render_cli_done: 

95 app._redraw(render_as_done=True) 

96 else: 

97 app.renderer.erase() 

98 

99 # Disable rendering. 

100 app._running_in_terminal = True 

101 

102 # Detach input. 

103 try: 

104 with app.input.detach(): 

105 with app.input.cooked_mode(): 

106 yield 

107 finally: 

108 # Redraw interface again. 

109 try: 

110 app._running_in_terminal = False 

111 app.renderer.reset() 

112 app._request_absolute_cursor_position() 

113 app._redraw() 

114 finally: 

115 # (Check for `.done()`, because it can be that this future was 

116 # cancelled.) 

117 if not new_run_in_terminal_f.done(): 

118 new_run_in_terminal_f.set_result(None)