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
[docs]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
[docs] 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("\n{}\n".format(kbi_msg))
return None
[docs] 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("\n{}\n".format(kbi_msg))
return None
[docs] 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()
[docs] 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
[docs] 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]