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
|
"""
Abstraction of CLI Input.
"""
from __future__ import annotations
from abc import ABCMeta, abstractmethod, abstractproperty
from contextlib import contextmanager
from typing import Callable, ContextManager, Generator
from prompt_toolkit.key_binding import KeyPress
__all__ = [
"Input",
"PipeInput",
"DummyInput",
]
class Input(metaclass=ABCMeta):
"""
Abstraction for any input.
An instance of this class can be given to the constructor of a
:class:`~prompt_toolkit.application.Application` and will also be
passed to the :class:`~prompt_toolkit.eventloop.base.EventLoop`.
"""
@abstractmethod
def fileno(self) -> int:
"""
Fileno for putting this in an event loop.
"""
@abstractmethod
def typeahead_hash(self) -> str:
"""
Identifier for storing type ahead key presses.
"""
@abstractmethod
def read_keys(self) -> list[KeyPress]:
"""
Return a list of Key objects which are read/parsed from the input.
"""
def flush_keys(self) -> list[KeyPress]:
"""
Flush the underlying parser. and return the pending keys.
(Used for vt100 input.)
"""
return []
def flush(self) -> None:
"The event loop can call this when the input has to be flushed."
pass
@abstractproperty
def closed(self) -> bool:
"Should be true when the input stream is closed."
return False
@abstractmethod
def raw_mode(self) -> ContextManager[None]:
"""
Context manager that turns the input into raw mode.
"""
@abstractmethod
def cooked_mode(self) -> ContextManager[None]:
"""
Context manager that turns the input into cooked mode.
"""
@abstractmethod
def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]:
"""
Return a context manager that makes this input active in the current
event loop.
"""
@abstractmethod
def detach(self) -> ContextManager[None]:
"""
Return a context manager that makes sure that this input is not active
in the current event loop.
"""
def close(self) -> None:
"Close input."
pass
class PipeInput(Input):
"""
Abstraction for pipe input.
"""
@abstractmethod
def send_bytes(self, data: bytes) -> None:
"""Feed byte string into the pipe"""
@abstractmethod
def send_text(self, data: str) -> None:
"""Feed a text string into the pipe"""
class DummyInput(Input):
"""
Input for use in a `DummyApplication`
If used in an actual application, it will make the application render
itself once and exit immediately, due to an `EOFError`.
"""
def fileno(self) -> int:
raise NotImplementedError
def typeahead_hash(self) -> str:
return "dummy-%s" % id(self)
def read_keys(self) -> list[KeyPress]:
return []
@property
def closed(self) -> bool:
# This needs to be true, so that the dummy input will trigger an
# `EOFError` immediately in the application.
return True
def raw_mode(self) -> ContextManager[None]:
return _dummy_context_manager()
def cooked_mode(self) -> ContextManager[None]:
return _dummy_context_manager()
def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]:
# Call the callback immediately once after attaching.
# This tells the callback to call `read_keys` and check the
# `input.closed` flag, after which it won't receive any keys, but knows
# that `EOFError` should be raised. This unblocks `read_from_input` in
# `application.py`.
input_ready_callback()
return _dummy_context_manager()
def detach(self) -> ContextManager[None]:
return _dummy_context_manager()
@contextmanager
def _dummy_context_manager() -> Generator[None, None, None]:
yield
|
