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

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 contextlib import asynccontextmanager 

9from typing import AsyncGenerator, Awaitable, Callable, TypeVar 

10 

11from prompt_toolkit.eventloop import run_in_executor_with_context 

12 

13from .current import get_app_or_none 

14 

15__all__ = [ 

16 "run_in_terminal", 

17 "in_terminal", 

18] 

19 

20_T = TypeVar("_T") 

21 

22 

23def run_in_terminal( 

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

25) -> Awaitable[_T]: 

26 """ 

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

28 

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

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

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

32 prompt. 

33 

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

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

36 manager directly. 

37 

38 :param func: The callable to execute. 

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

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

41 erase the interface first. 

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

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

44 

45 :returns: A `Future`. 

46 """ 

47 

48 async def run() -> _T: 

49 async with in_terminal(render_cli_done=render_cli_done): 

50 if in_executor: 

51 return await run_in_executor_with_context(func) 

52 else: 

53 return func() 

54 

55 return ensure_future(run()) 

56 

57 

58@asynccontextmanager 

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

60 """ 

61 Asynchronous context manager that suspends the current application and runs 

62 the body in the terminal. 

63 

64 .. code:: 

65 

66 async def f(): 

67 async with in_terminal(): 

68 call_some_function() 

69 await call_some_async_function() 

70 """ 

71 app = get_app_or_none() 

72 if app is None or not app._is_running: 

73 yield 

74 return 

75 

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

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

78 previous_run_in_terminal_f = app._running_in_terminal_f 

79 new_run_in_terminal_f: Future[None] = Future() 

80 app._running_in_terminal_f = new_run_in_terminal_f 

81 

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

83 if previous_run_in_terminal_f is not None: 

84 await previous_run_in_terminal_f 

85 

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

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

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

89 if app.output.responds_to_cpr: 

90 await app.renderer.wait_for_cpr_responses() 

91 

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

93 if render_cli_done: 

94 app._redraw(render_as_done=True) 

95 else: 

96 app.renderer.erase() 

97 

98 # Disable rendering. 

99 app._running_in_terminal = True 

100 

101 # Detach input. 

102 try: 

103 with app.input.detach(): 

104 with app.input.cooked_mode(): 

105 yield 

106 finally: 

107 # Redraw interface again. 

108 try: 

109 app._running_in_terminal = False 

110 app.renderer.reset() 

111 app._request_absolute_cursor_position() 

112 app._redraw() 

113 finally: 

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

115 # cancelled.) 

116 if not new_run_in_terminal_f.done(): 

117 new_run_in_terminal_f.set_result(None)