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