1"""
2Abstraction of CLI Input.
3"""
4
5from __future__ import annotations
6
7from abc import ABCMeta, abstractmethod
8from collections.abc import Callable, Generator
9from contextlib import AbstractContextManager, contextmanager
10
11from prompt_toolkit.key_binding import KeyPress
12
13__all__ = [
14 "Input",
15 "PipeInput",
16 "DummyInput",
17]
18
19
20class Input(metaclass=ABCMeta):
21 """
22 Abstraction for any input.
23
24 An instance of this class can be given to the constructor of a
25 :class:`~prompt_toolkit.application.Application` and will also be
26 passed to the :class:`~prompt_toolkit.eventloop.base.EventLoop`.
27 """
28
29 @abstractmethod
30 def fileno(self) -> int:
31 """
32 Fileno for putting this in an event loop.
33 """
34
35 @abstractmethod
36 def typeahead_hash(self) -> str:
37 """
38 Identifier for storing type ahead key presses.
39 """
40
41 @abstractmethod
42 def read_keys(self) -> list[KeyPress]:
43 """
44 Return a list of Key objects which are read/parsed from the input.
45 """
46
47 def flush_keys(self) -> list[KeyPress]:
48 """
49 Flush the underlying parser. and return the pending keys.
50 (Used for vt100 input.)
51 """
52 return []
53
54 def flush(self) -> None:
55 "The event loop can call this when the input has to be flushed."
56 pass
57
58 @property
59 @abstractmethod
60 def closed(self) -> bool:
61 "Should be true when the input stream is closed."
62 return False
63
64 @abstractmethod
65 def raw_mode(self) -> AbstractContextManager[None]:
66 """
67 Context manager that turns the input into raw mode.
68 """
69
70 @abstractmethod
71 def cooked_mode(self) -> AbstractContextManager[None]:
72 """
73 Context manager that turns the input into cooked mode.
74 """
75
76 @abstractmethod
77 def attach(
78 self, input_ready_callback: Callable[[], None]
79 ) -> AbstractContextManager[None]:
80 """
81 Return a context manager that makes this input active in the current
82 event loop.
83 """
84
85 @abstractmethod
86 def detach(self) -> AbstractContextManager[None]:
87 """
88 Return a context manager that makes sure that this input is not active
89 in the current event loop.
90 """
91
92 def close(self) -> None:
93 "Close input."
94 pass
95
96
97class PipeInput(Input):
98 """
99 Abstraction for pipe input.
100 """
101
102 @abstractmethod
103 def send_bytes(self, data: bytes) -> None:
104 """Feed byte string into the pipe"""
105
106 @abstractmethod
107 def send_text(self, data: str) -> None:
108 """Feed a text string into the pipe"""
109
110
111class DummyInput(Input):
112 """
113 Input for use in a `DummyApplication`
114
115 If used in an actual application, it will make the application render
116 itself once and exit immediately, due to an `EOFError`.
117 """
118
119 def fileno(self) -> int:
120 raise NotImplementedError
121
122 def typeahead_hash(self) -> str:
123 return f"dummy-{id(self)}"
124
125 def read_keys(self) -> list[KeyPress]:
126 return []
127
128 @property
129 def closed(self) -> bool:
130 # This needs to be true, so that the dummy input will trigger an
131 # `EOFError` immediately in the application.
132 return True
133
134 def raw_mode(self) -> AbstractContextManager[None]:
135 return _dummy_context_manager()
136
137 def cooked_mode(self) -> AbstractContextManager[None]:
138 return _dummy_context_manager()
139
140 def attach(
141 self, input_ready_callback: Callable[[], None]
142 ) -> AbstractContextManager[None]:
143 # Call the callback immediately once after attaching.
144 # This tells the callback to call `read_keys` and check the
145 # `input.closed` flag, after which it won't receive any keys, but knows
146 # that `EOFError` should be raised. This unblocks `read_from_input` in
147 # `application.py`.
148 input_ready_callback()
149
150 return _dummy_context_manager()
151
152 def detach(self) -> AbstractContextManager[None]:
153 return _dummy_context_manager()
154
155
156@contextmanager
157def _dummy_context_manager() -> Generator[None, None, None]:
158 yield