1"""
2`Fish-style <http://fishshell.com/>`_ like auto-suggestion.
3
4While a user types input in a certain buffer, suggestions are generated
5(asynchronously.) Usually, they are displayed after the input. When the cursor
6presses the right arrow and the cursor is at the end of the input, the
7suggestion will be inserted.
8
9If you want the auto suggestions to be asynchronous (in a background thread),
10because they take too much time, and could potentially block the event loop,
11then wrap the :class:`.AutoSuggest` instance into a
12:class:`.ThreadedAutoSuggest`.
13"""
14
15from __future__ import annotations
16
17from abc import ABCMeta, abstractmethod
18from typing import TYPE_CHECKING, Callable
19
20from prompt_toolkit.eventloop import run_in_executor_with_context
21
22from .document import Document
23from .filters import Filter, to_filter
24
25if TYPE_CHECKING:
26 from .buffer import Buffer
27
28__all__ = [
29 "Suggestion",
30 "AutoSuggest",
31 "ThreadedAutoSuggest",
32 "DummyAutoSuggest",
33 "AutoSuggestFromHistory",
34 "ConditionalAutoSuggest",
35 "DynamicAutoSuggest",
36]
37
38
39class Suggestion:
40 """
41 Suggestion returned by an auto-suggest algorithm.
42
43 :param text: The suggestion text.
44 """
45
46 def __init__(self, text: str) -> None:
47 self.text = text
48
49 def __repr__(self) -> str:
50 return f"Suggestion({self.text})"
51
52
53class AutoSuggest(metaclass=ABCMeta):
54 """
55 Base class for auto suggestion implementations.
56 """
57
58 @abstractmethod
59 def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None:
60 """
61 Return `None` or a :class:`.Suggestion` instance.
62
63 We receive both :class:`~prompt_toolkit.buffer.Buffer` and
64 :class:`~prompt_toolkit.document.Document`. The reason is that auto
65 suggestions are retrieved asynchronously. (Like completions.) The
66 buffer text could be changed in the meantime, but ``document`` contains
67 the buffer document like it was at the start of the auto suggestion
68 call. So, from here, don't access ``buffer.text``, but use
69 ``document.text`` instead.
70
71 :param buffer: The :class:`~prompt_toolkit.buffer.Buffer` instance.
72 :param document: The :class:`~prompt_toolkit.document.Document` instance.
73 """
74
75 async def get_suggestion_async(
76 self, buff: Buffer, document: Document
77 ) -> Suggestion | None:
78 """
79 Return a :class:`.Future` which is set when the suggestions are ready.
80 This function can be overloaded in order to provide an asynchronous
81 implementation.
82 """
83 return self.get_suggestion(buff, document)
84
85
86class ThreadedAutoSuggest(AutoSuggest):
87 """
88 Wrapper that runs auto suggestions in a thread.
89 (Use this to prevent the user interface from becoming unresponsive if the
90 generation of suggestions takes too much time.)
91 """
92
93 def __init__(self, auto_suggest: AutoSuggest) -> None:
94 self.auto_suggest = auto_suggest
95
96 def get_suggestion(self, buff: Buffer, document: Document) -> Suggestion | None:
97 return self.auto_suggest.get_suggestion(buff, document)
98
99 async def get_suggestion_async(
100 self, buff: Buffer, document: Document
101 ) -> Suggestion | None:
102 """
103 Run the `get_suggestion` function in a thread.
104 """
105
106 def run_get_suggestion_thread() -> Suggestion | None:
107 return self.get_suggestion(buff, document)
108
109 return await run_in_executor_with_context(run_get_suggestion_thread)
110
111
112class DummyAutoSuggest(AutoSuggest):
113 """
114 AutoSuggest class that doesn't return any suggestion.
115 """
116
117 def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None:
118 return None # No suggestion
119
120
121class AutoSuggestFromHistory(AutoSuggest):
122 """
123 Give suggestions based on the lines in the history.
124 """
125
126 def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None:
127 history = buffer.history
128
129 # Consider only the last line for the suggestion.
130 text = document.text.rsplit("\n", 1)[-1]
131
132 # Only create a suggestion when this is not an empty line.
133 if text.strip():
134 # Find first matching line in history.
135 for string in reversed(list(history.get_strings())):
136 for line in reversed(string.splitlines()):
137 if line.startswith(text):
138 return Suggestion(line[len(text) :])
139
140 return None
141
142
143class ConditionalAutoSuggest(AutoSuggest):
144 """
145 Auto suggest that can be turned on and of according to a certain condition.
146 """
147
148 def __init__(self, auto_suggest: AutoSuggest, filter: bool | Filter) -> None:
149 self.auto_suggest = auto_suggest
150 self.filter = to_filter(filter)
151
152 def get_suggestion(self, buffer: Buffer, document: Document) -> Suggestion | None:
153 if self.filter():
154 return self.auto_suggest.get_suggestion(buffer, document)
155
156 return None
157
158
159class DynamicAutoSuggest(AutoSuggest):
160 """
161 Validator class that can dynamically returns any Validator.
162
163 :param get_validator: Callable that returns a :class:`.Validator` instance.
164 """
165
166 def __init__(self, get_auto_suggest: Callable[[], AutoSuggest | None]) -> None:
167 self.get_auto_suggest = get_auto_suggest
168
169 def get_suggestion(self, buff: Buffer, document: Document) -> Suggestion | None:
170 auto_suggest = self.get_auto_suggest() or DummyAutoSuggest()
171 return auto_suggest.get_suggestion(buff, document)
172
173 async def get_suggestion_async(
174 self, buff: Buffer, document: Document
175 ) -> Suggestion | None:
176 auto_suggest = self.get_auto_suggest() or DummyAutoSuggest()
177 return await auto_suggest.get_suggestion_async(buff, document)