1"""
2Input validation for a `Buffer`.
3(Validators will be called before accepting input.)
4"""
5
6from __future__ import annotations
7
8from abc import ABCMeta, abstractmethod
9from typing import Callable
10
11from prompt_toolkit.eventloop import run_in_executor_with_context
12
13from .document import Document
14from .filters import FilterOrBool, to_filter
15
16__all__ = [
17 "ConditionalValidator",
18 "ValidationError",
19 "Validator",
20 "ThreadedValidator",
21 "DummyValidator",
22 "DynamicValidator",
23]
24
25
26class ValidationError(Exception):
27 """
28 Error raised by :meth:`.Validator.validate`.
29
30 :param cursor_position: The cursor position where the error occurred.
31 :param message: Text.
32 """
33
34 def __init__(self, cursor_position: int = 0, message: str = "") -> None:
35 super().__init__(message)
36 self.cursor_position = cursor_position
37 self.message = message
38
39 def __repr__(self) -> str:
40 return f"{self.__class__.__name__}(cursor_position={self.cursor_position!r}, message={self.message!r})"
41
42
43class Validator(metaclass=ABCMeta):
44 """
45 Abstract base class for an input validator.
46
47 A validator is typically created in one of the following two ways:
48
49 - Either by overriding this class and implementing the `validate` method.
50 - Or by passing a callable to `Validator.from_callable`.
51
52 If the validation takes some time and needs to happen in a background
53 thread, this can be wrapped in a :class:`.ThreadedValidator`.
54 """
55
56 @abstractmethod
57 def validate(self, document: Document) -> None:
58 """
59 Validate the input.
60 If invalid, this should raise a :class:`.ValidationError`.
61
62 :param document: :class:`~prompt_toolkit.document.Document` instance.
63 """
64 pass
65
66 async def validate_async(self, document: Document) -> None:
67 """
68 Return a `Future` which is set when the validation is ready.
69 This function can be overloaded in order to provide an asynchronous
70 implementation.
71 """
72 try:
73 self.validate(document)
74 except ValidationError:
75 raise
76
77 @classmethod
78 def from_callable(
79 cls,
80 validate_func: Callable[[str], bool],
81 error_message: str = "Invalid input",
82 move_cursor_to_end: bool = False,
83 ) -> Validator:
84 """
85 Create a validator from a simple validate callable. E.g.:
86
87 .. code:: python
88
89 def is_valid(text):
90 return text in ['hello', 'world']
91 Validator.from_callable(is_valid, error_message='Invalid input')
92
93 :param validate_func: Callable that takes the input string, and returns
94 `True` if the input is valid input.
95 :param error_message: Message to be displayed if the input is invalid.
96 :param move_cursor_to_end: Move the cursor to the end of the input, if
97 the input is invalid.
98 """
99 return _ValidatorFromCallable(validate_func, error_message, move_cursor_to_end)
100
101
102class _ValidatorFromCallable(Validator):
103 """
104 Validate input from a simple callable.
105 """
106
107 def __init__(
108 self, func: Callable[[str], bool], error_message: str, move_cursor_to_end: bool
109 ) -> None:
110 self.func = func
111 self.error_message = error_message
112 self.move_cursor_to_end = move_cursor_to_end
113
114 def __repr__(self) -> str:
115 return f"Validator.from_callable({self.func!r})"
116
117 def validate(self, document: Document) -> None:
118 if not self.func(document.text):
119 if self.move_cursor_to_end:
120 index = len(document.text)
121 else:
122 index = 0
123
124 raise ValidationError(cursor_position=index, message=self.error_message)
125
126
127class ThreadedValidator(Validator):
128 """
129 Wrapper that runs input validation in a thread.
130 (Use this to prevent the user interface from becoming unresponsive if the
131 input validation takes too much time.)
132 """
133
134 def __init__(self, validator: Validator) -> None:
135 self.validator = validator
136
137 def validate(self, document: Document) -> None:
138 self.validator.validate(document)
139
140 async def validate_async(self, document: Document) -> None:
141 """
142 Run the `validate` function in a thread.
143 """
144
145 def run_validation_thread() -> None:
146 return self.validate(document)
147
148 await run_in_executor_with_context(run_validation_thread)
149
150
151class DummyValidator(Validator):
152 """
153 Validator class that accepts any input.
154 """
155
156 def validate(self, document: Document) -> None:
157 pass # Don't raise any exception.
158
159
160class ConditionalValidator(Validator):
161 """
162 Validator that can be switched on/off according to
163 a filter. (This wraps around another validator.)
164 """
165
166 def __init__(self, validator: Validator, filter: FilterOrBool) -> None:
167 self.validator = validator
168 self.filter = to_filter(filter)
169
170 def validate(self, document: Document) -> None:
171 # Call the validator only if the filter is active.
172 if self.filter():
173 self.validator.validate(document)
174
175
176class DynamicValidator(Validator):
177 """
178 Validator class that can dynamically returns any Validator.
179
180 :param get_validator: Callable that returns a :class:`.Validator` instance.
181 """
182
183 def __init__(self, get_validator: Callable[[], Validator | None]) -> None:
184 self.get_validator = get_validator
185
186 def validate(self, document: Document) -> None:
187 validator = self.get_validator() or DummyValidator()
188 validator.validate(document)
189
190 async def validate_async(self, document: Document) -> None:
191 validator = self.get_validator() or DummyValidator()
192 await validator.validate_async(document)