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