1"""
2Abstraction of CLI Input.
3"""
4
5from __future__ import annotations
6
7from abc import ABCMeta, abstractmethod
8from contextlib import contextmanager
9from typing import Callable, ContextManager, Generator
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) -> ContextManager[None]:
66 """
67 Context manager that turns the input into raw mode.
68 """
69
70 @abstractmethod
71 def cooked_mode(self) -> ContextManager[None]:
72 """
73 Context manager that turns the input into cooked mode.
74 """
75
76 @abstractmethod
77 def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]:
78 """
79 Return a context manager that makes this input active in the current
80 event loop.
81 """
82
83 @abstractmethod
84 def detach(self) -> ContextManager[None]:
85 """
86 Return a context manager that makes sure that this input is not active
87 in the current event loop.
88 """
89
90 def close(self) -> None:
91 "Close input."
92 pass
93
94
95class PipeInput(Input):
96 """
97 Abstraction for pipe input.
98 """
99
100 @abstractmethod
101 def send_bytes(self, data: bytes) -> None:
102 """Feed byte string into the pipe"""
103
104 @abstractmethod
105 def send_text(self, data: str) -> None:
106 """Feed a text string into the pipe"""
107
108
109class DummyInput(Input):
110 """
111 Input for use in a `DummyApplication`
112
113 If used in an actual application, it will make the application render
114 itself once and exit immediately, due to an `EOFError`.
115 """
116
117 def fileno(self) -> int:
118 raise NotImplementedError
119
120 def typeahead_hash(self) -> str:
121 return f"dummy-{id(self)}"
122
123 def read_keys(self) -> list[KeyPress]:
124 return []
125
126 @property
127 def closed(self) -> bool:
128 # This needs to be true, so that the dummy input will trigger an
129 # `EOFError` immediately in the application.
130 return True
131
132 def raw_mode(self) -> ContextManager[None]:
133 return _dummy_context_manager()
134
135 def cooked_mode(self) -> ContextManager[None]:
136 return _dummy_context_manager()
137
138 def attach(self, input_ready_callback: Callable[[], None]) -> ContextManager[None]:
139 # Call the callback immediately once after attaching.
140 # This tells the callback to call `read_keys` and check the
141 # `input.closed` flag, after which it won't receive any keys, but knows
142 # that `EOFError` should be raised. This unblocks `read_from_input` in
143 # `application.py`.
144 input_ready_callback()
145
146 return _dummy_context_manager()
147
148 def detach(self) -> ContextManager[None]:
149 return _dummy_context_manager()
150
151
152@contextmanager
153def _dummy_context_manager() -> Generator[None, None, None]:
154 yield