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)