base.py 3.9 KB

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