|
|
"""Contains the base class :class:`.BaseSimplePrompt`.""" |
|
|
import os |
|
|
import re |
|
|
from abc import ABC, abstractmethod |
|
|
from typing import ( |
|
|
TYPE_CHECKING, |
|
|
Any, |
|
|
Callable, |
|
|
Dict, |
|
|
List, |
|
|
Optional, |
|
|
Tuple, |
|
|
Union, |
|
|
cast, |
|
|
) |
|
|
|
|
|
from prompt_toolkit.enums import EditingMode |
|
|
from prompt_toolkit.filters.base import Condition, FilterOrBool |
|
|
from prompt_toolkit.key_binding.key_bindings import KeyBindings, KeyHandlerCallable |
|
|
from prompt_toolkit.keys import Keys |
|
|
from prompt_toolkit.styles.style import Style |
|
|
from prompt_toolkit.validation import Validator |
|
|
|
|
|
from InquirerPy.enum import INQUIRERPY_KEYBOARD_INTERRUPT |
|
|
from InquirerPy.exceptions import RequiredKeyNotFound |
|
|
from InquirerPy.utils import ( |
|
|
InquirerPyMessage, |
|
|
InquirerPySessionResult, |
|
|
InquirerPyStyle, |
|
|
InquirerPyValidate, |
|
|
get_style, |
|
|
) |
|
|
|
|
|
if TYPE_CHECKING: |
|
|
from prompt_toolkit.key_binding.key_processor import KeyPressEvent |
|
|
|
|
|
|
|
|
class BaseSimplePrompt(ABC): |
|
|
"""The base class to create a simple terminal input prompt. |
|
|
|
|
|
Note: |
|
|
No actual :class:`~prompt_toolkit.application.Application` is created by this class. |
|
|
This class only creates some common interface and attributes that can be easily used |
|
|
by `prompt_toolkit`. |
|
|
|
|
|
To have a functional prompt, you'll at least have to implement the :meth:`.BaseSimplePrompt._run` |
|
|
and :meth:`.BaseSimplePrompt._get_prompt_message`. |
|
|
|
|
|
See Also: |
|
|
:class:`~InquirerPy.prompts.input.InputPrompt` |
|
|
""" |
|
|
|
|
|
def __init__( |
|
|
self, |
|
|
message: InquirerPyMessage, |
|
|
style: Optional[InquirerPyStyle] = None, |
|
|
vi_mode: bool = False, |
|
|
qmark: str = "?", |
|
|
amark: str = "?", |
|
|
instruction: str = "", |
|
|
validate: Optional[InquirerPyValidate] = None, |
|
|
invalid_message: str = "Invalid input", |
|
|
transformer: Optional[Callable[[Any], Any]] = None, |
|
|
filter: Optional[Callable[[Any], Any]] = None, |
|
|
default: Any = "", |
|
|
wrap_lines: bool = True, |
|
|
raise_keyboard_interrupt: bool = True, |
|
|
mandatory: bool = True, |
|
|
mandatory_message: str = "Mandatory prompt", |
|
|
session_result: Optional[InquirerPySessionResult] = None, |
|
|
) -> None: |
|
|
self._mandatory = mandatory |
|
|
self._mandatory_message = mandatory_message |
|
|
self._result = session_result or {} |
|
|
self._message = ( |
|
|
message |
|
|
if not isinstance(message, Callable) |
|
|
else cast(Callable, message)(self._result) |
|
|
) |
|
|
self._instruction = instruction |
|
|
self._default = ( |
|
|
default if not isinstance(default, Callable) else default(self._result) |
|
|
) |
|
|
self._style = Style.from_dict(style.dict if style else get_style().dict) |
|
|
self._qmark = qmark |
|
|
self._amark = amark |
|
|
self._status = {"answered": False, "result": None, "skipped": False} |
|
|
self._kb = KeyBindings() |
|
|
self._lexer = "class:input" |
|
|
self._transformer = transformer |
|
|
self._filter = filter |
|
|
self._wrap_lines = wrap_lines |
|
|
self._editing_mode = ( |
|
|
EditingMode.VI |
|
|
if vi_mode or bool(os.getenv("INQUIRERPY_VI_MODE", False)) |
|
|
else EditingMode.EMACS |
|
|
) |
|
|
if isinstance(validate, Validator): |
|
|
self._validator = validate |
|
|
else: |
|
|
self._validator = Validator.from_callable( |
|
|
validate if validate else lambda _: True, |
|
|
invalid_message, |
|
|
move_cursor_to_end=True, |
|
|
) |
|
|
self._raise_kbi = not os.getenv( |
|
|
"INQUIRERPY_NO_RAISE_KBI", not raise_keyboard_interrupt |
|
|
) |
|
|
self._is_rasing_kbi = Condition(lambda: self._raise_kbi) |
|
|
|
|
|
self._kb_maps = { |
|
|
"answer": [{"key": Keys.Enter}], |
|
|
"interrupt": [ |
|
|
{"key": "c-c", "filter": self._is_rasing_kbi}, |
|
|
{"key": "c-d", "filter": ~self._is_rasing_kbi}, |
|
|
], |
|
|
"skip": [{"key": "c-z"}, {"key": "c-c", "filter": ~self._is_rasing_kbi}], |
|
|
} |
|
|
self._kb_func_lookup = { |
|
|
"answer": [{"func": self._handle_enter}], |
|
|
"interrupt": [{"func": self._handle_interrupt}], |
|
|
"skip": [{"func": self._handle_skip}], |
|
|
} |
|
|
|
|
|
def _keybinding_factory(self): |
|
|
"""Register all keybindings in `self._kb_maps`. |
|
|
|
|
|
It's required to call this function at the end of prompt constructor if |
|
|
it inherits from :class:`~InquirerPy.base.simple.BaseSimplePrompt` or |
|
|
:class:`~InquirerPy.base.complex.BaseComplexPrompt`. |
|
|
""" |
|
|
|
|
|
def _factory(keys, filter, action): |
|
|
if action not in self.kb_func_lookup: |
|
|
raise RequiredKeyNotFound(f"keybinding action {action} not found") |
|
|
if not isinstance(keys, list): |
|
|
keys = [keys] |
|
|
|
|
|
@self.register_kb(*keys, filter=filter) |
|
|
def _(event): |
|
|
for method in self.kb_func_lookup[action]: |
|
|
method["func"](event, *method.get("args", [])) |
|
|
|
|
|
for key, item in self.kb_maps.items(): |
|
|
if not isinstance(item, list): |
|
|
item = [item] |
|
|
for kb in item: |
|
|
_factory(kb["key"], kb.get("filter", Condition(lambda: True)), key) |
|
|
|
|
|
@abstractmethod |
|
|
def _set_error(self, message: str) -> None: |
|
|
"""Set the error message for the prompt. |
|
|
|
|
|
Args: |
|
|
message: Error message to set. |
|
|
""" |
|
|
pass |
|
|
|
|
|
def _handle_skip(self, event: Optional["KeyPressEvent"]) -> None: |
|
|
"""Handle the event when attempting to skip a prompt. |
|
|
|
|
|
Skip the prompt if the `_mandatory` field is False, otherwise |
|
|
show an error message that the prompt cannot be skipped. |
|
|
""" |
|
|
if not self._mandatory: |
|
|
self.status["answered"] = True |
|
|
self.status["skipped"] = True |
|
|
self.status["result"] = None |
|
|
if event: |
|
|
event.app.exit(result=None) |
|
|
else: |
|
|
self._set_error(message=self._mandatory_message) |
|
|
|
|
|
def _handle_interrupt(self, event: Optional["KeyPressEvent"]) -> None: |
|
|
"""Handle the event when a KeyboardInterrupt signal is sent.""" |
|
|
self.status["answered"] = True |
|
|
self.status["result"] = INQUIRERPY_KEYBOARD_INTERRUPT |
|
|
self.status["skipped"] = True |
|
|
if event: |
|
|
event.app.exit(result=INQUIRERPY_KEYBOARD_INTERRUPT) |
|
|
|
|
|
@abstractmethod |
|
|
def _handle_enter(self, event: Optional["KeyPressEvent"]) -> None: |
|
|
"""Handle the event when user attempt to answer the question.""" |
|
|
pass |
|
|
|
|
|
@property |
|
|
def status(self) -> Dict[str, Any]: |
|
|
"""Dict[str, Any]: Get current prompt status. |
|
|
|
|
|
The status contains 3 keys: "answered" and "result". |
|
|
answered: If the current prompt is answered. |
|
|
result: The result of the user answer. |
|
|
skipped: If the prompt is skipped. |
|
|
""" |
|
|
return self._status |
|
|
|
|
|
@status.setter |
|
|
def status(self, value) -> None: |
|
|
self._status = value |
|
|
|
|
|
def register_kb( |
|
|
self, *keys: Union[Keys, str], filter: FilterOrBool = True, **kwargs |
|
|
) -> Callable[[KeyHandlerCallable], KeyHandlerCallable]: |
|
|
"""Keybinding registration decorator. |
|
|
|
|
|
This decorator wraps around the :meth:`prompt_toolkit.key_binding.KeyBindings.add` with |
|
|
added feature to process `alt` realted keybindings. |
|
|
|
|
|
By default, `prompt_toolkit` doesn't process `alt` related keybindings, |
|
|
it requires `alt-ANY` to `escape` + `ANY`. |
|
|
|
|
|
Args: |
|
|
keys: The keys to bind that can trigger the function. |
|
|
filter: :class:`~prompt_toolkit.filter.Condition` to indicate if this keybinding should be active. |
|
|
|
|
|
Returns: |
|
|
A decorator that should be applied to the function thats intended to be active when the keys |
|
|
are pressed. |
|
|
|
|
|
Examples: |
|
|
>>> @self.register_kb("alt-j") |
|
|
... def test(event): |
|
|
... pass |
|
|
""" |
|
|
alt_pattern = re.compile(r"^alt-(.*)") |
|
|
|
|
|
def decorator(func: KeyHandlerCallable) -> KeyHandlerCallable: |
|
|
formatted_keys = [] |
|
|
for key in keys: |
|
|
match = alt_pattern.match(key) |
|
|
if match: |
|
|
formatted_keys.append("escape") |
|
|
formatted_keys.append(match.group(1)) |
|
|
else: |
|
|
formatted_keys.append(key) |
|
|
|
|
|
@self._kb.add(*formatted_keys, filter=filter, **kwargs) |
|
|
def executable(event) -> None: |
|
|
func(event) |
|
|
|
|
|
return executable |
|
|
|
|
|
return decorator |
|
|
|
|
|
@abstractmethod |
|
|
def _get_prompt_message( |
|
|
self, pre_answer: Tuple[str, str], post_answer: Tuple[str, str] |
|
|
) -> List[Tuple[str, str]]: |
|
|
"""Get the question message in formatted text form to display in the prompt. |
|
|
|
|
|
This function is mainly used to render the question message dynamically based |
|
|
on the current status (answered or not answered) of the prompt. |
|
|
|
|
|
Note: |
|
|
The function requires implementation when inheriting :class:`.BaseSimplePrompt`. |
|
|
You should call `super()._get_prompt_message(pre_answer, post_answer)` in |
|
|
the implemented `_get_prompt_message`. |
|
|
|
|
|
Args: |
|
|
pre_answer: The message to display before the question is answered. |
|
|
post_answer: The information to display after the question is answered. |
|
|
|
|
|
Returns: |
|
|
Formatted text in list of tuple format. |
|
|
""" |
|
|
display_message = [] |
|
|
if self.status["skipped"]: |
|
|
display_message.append(("class:skipped", self._qmark)) |
|
|
display_message.append( |
|
|
("class:skipped", "%s%s " % (" " if self._qmark else "", self._message)) |
|
|
) |
|
|
elif self.status["answered"]: |
|
|
display_message.append(("class:answermark", self._amark)) |
|
|
display_message.append( |
|
|
( |
|
|
"class:answered_question", |
|
|
"%s%s" % (" " if self._amark else "", self._message), |
|
|
) |
|
|
) |
|
|
display_message.append( |
|
|
post_answer |
|
|
if not self._transformer |
|
|
else ( |
|
|
"class:answer", |
|
|
" %s" % self._transformer(self.status["result"]), |
|
|
) |
|
|
) |
|
|
else: |
|
|
display_message.append(("class:questionmark", self._qmark)) |
|
|
display_message.append( |
|
|
( |
|
|
"class:question", |
|
|
"%s%s" % (" " if self._qmark else "", self._message), |
|
|
) |
|
|
) |
|
|
display_message.append(pre_answer) |
|
|
return display_message |
|
|
|
|
|
@abstractmethod |
|
|
def _run(self) -> Any: |
|
|
"""Abstractmethod to enforce a run function is implemented. |
|
|
|
|
|
All prompt instance requires a `_run` call to initialise and run an instance of |
|
|
`PromptSession` or `Application`. |
|
|
""" |
|
|
pass |
|
|
|
|
|
@abstractmethod |
|
|
async def _run_async(self) -> Any: |
|
|
"""Abstractmethod to enforce a run function is implemented. |
|
|
|
|
|
All prompt instance requires a `_run_async` call to initialise and run an instance of |
|
|
`PromptSession` or `Application`. |
|
|
""" |
|
|
pass |
|
|
|
|
|
def execute(self, raise_keyboard_interrupt: Optional[bool] = None) -> Any: |
|
|
"""Run the prompt and get the result. |
|
|
|
|
|
Args: |
|
|
raise_keyboard_interrupt: **Deprecated**. Set this parameter on the prompt initialisation instead. |
|
|
|
|
|
Returns: |
|
|
Value of the user answer. Types varies depending on the prompt. |
|
|
|
|
|
Raises: |
|
|
KeyboardInterrupt: When `ctrl-c` is pressed and `raise_keyboard_interrupt` is True. |
|
|
""" |
|
|
result = self._run() |
|
|
if raise_keyboard_interrupt is not None: |
|
|
self._raise_kbi = not os.getenv( |
|
|
"INQUIRERPY_NO_RAISE_KBI", not raise_keyboard_interrupt |
|
|
) |
|
|
if result == INQUIRERPY_KEYBOARD_INTERRUPT: |
|
|
raise KeyboardInterrupt |
|
|
if not self._filter: |
|
|
return result |
|
|
return self._filter(result) |
|
|
|
|
|
async def execute_async(self) -> None: |
|
|
"""Run the prompt asynchronously and get the result. |
|
|
|
|
|
Returns: |
|
|
Value of the user answer. Types varies depending on the prompt. |
|
|
|
|
|
Raises: |
|
|
KeyboardInterrupt: When `ctrl-c` is pressed and `raise_keyboard_interrupt` is True. |
|
|
""" |
|
|
result = await self._run_async() |
|
|
if result == INQUIRERPY_KEYBOARD_INTERRUPT: |
|
|
raise KeyboardInterrupt |
|
|
if not self._filter: |
|
|
return result |
|
|
return self._filter(result) |
|
|
|
|
|
@property |
|
|
def instruction(self) -> str: |
|
|
"""str: Instruction to display next to question.""" |
|
|
return self._instruction |
|
|
|
|
|
@property |
|
|
def kb_maps(self) -> Dict[str, Any]: |
|
|
"""Dict[str, Any]: Keybinding mappings.""" |
|
|
return self._kb_maps |
|
|
|
|
|
@kb_maps.setter |
|
|
def kb_maps(self, value: Dict[str, Any]) -> None: |
|
|
self._kb_maps = {**self._kb_maps, **value} |
|
|
|
|
|
@property |
|
|
def kb_func_lookup(self) -> Dict[str, Any]: |
|
|
"""Dict[str, Any]: Keybinding function lookup mappings..""" |
|
|
return self._kb_func_lookup |
|
|
|
|
|
@kb_func_lookup.setter |
|
|
def kb_func_lookup(self, value: Dict[str, Any]) -> None: |
|
|
self._kb_func_lookup = {**self._kb_func_lookup, **value} |
|
|
|
