1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
import sys
from typing import Any
import prompt_toolkit.patch_stdout
from prompt_toolkit import Application
from questionary import utils
from questionary.constants import DEFAULT_KBI_MESSAGE
class Question:
"""A question to be prompted.
This is an internal class. Questions should be created using the
predefined questions (e.g. text or password)."""
application: "Application[Any]"
should_skip_question: bool
default: Any
def __init__(self, application: "Application[Any]") -> None:
self.application = application
self.should_skip_question = False
self.default = None
async def ask_async(
self, patch_stdout: bool = False, kbi_msg: str = DEFAULT_KBI_MESSAGE
) -> Any:
"""Ask the question using asyncio and return user response.
Args:
patch_stdout: Ensure that the prompt renders correctly if other threads
are printing to stdout.
kbi_msg: The message to be printed on a keyboard interrupt.
Returns:
`Any`: The answer from the question.
"""
try:
sys.stdout.flush()
return await self.unsafe_ask_async(patch_stdout)
except KeyboardInterrupt:
print("{}".format(kbi_msg))
return None
def ask(
self, patch_stdout: bool = False, kbi_msg: str = DEFAULT_KBI_MESSAGE
) -> Any:
"""Ask the question synchronously and return user response.
Args:
patch_stdout: Ensure that the prompt renders correctly if other threads
are printing to stdout.
kbi_msg: The message to be printed on a keyboard interrupt.
Returns:
`Any`: The answer from the question.
"""
try:
return self.unsafe_ask(patch_stdout)
except KeyboardInterrupt:
print("{}".format(kbi_msg))
return None
def unsafe_ask(self, patch_stdout: bool = False) -> Any:
"""Ask the question synchronously and return user response.
Does not catch keyboard interrupts.
Args:
patch_stdout: Ensure that the prompt renders correctly if other threads
are printing to stdout.
Returns:
`Any`: The answer from the question.
"""
if self.should_skip_question:
return self.default
if patch_stdout:
with prompt_toolkit.patch_stdout.patch_stdout():
return self.application.run()
else:
return self.application.run()
def skip_if(self, condition: bool, default: Any = None) -> "Question":
"""Skip the question if flag is set and return the default instead.
Args:
condition: A conditional boolean value.
default: The default value to return.
Returns:
:class:`Question`: `self`.
"""
self.should_skip_question = condition
self.default = default
return self
async def unsafe_ask_async(self, patch_stdout: bool = False) -> Any:
"""Ask the question using asyncio and return user response.
Does not catch keyboard interrupts.
Args:
patch_stdout: Ensure that the prompt renders correctly if other threads
are printing to stdout.
Returns:
`Any`: The answer from the question.
"""
if self.should_skip_question:
return self.default
if not utils.ACTIVATED_ASYNC_MODE:
await utils.activate_prompt_toolkit_async_mode()
if patch_stdout:
with prompt_toolkit.patch_stdout.patch_stdout():
r = self.application.run_async()
else:
r = self.application.run_async()
if utils.is_prompt_toolkit_3():
return await r
else:
return await r.to_asyncio_future() # type: ignore[attr-defined]
|
