gh-119842: Honor PyOS_InputHook in the new REPL (GH-119843)

Signed-off-by: Pablo Galindo <[email protected]>
Co-authored-by: Łukasz Langa <[email protected]>
Co-authored-by: Michael Droettboom <[email protected]>

Lib/_pyrepl/console.py

@@ -33,6 +33,7 @@
 
 if TYPE_CHECKING:
     from typing import IO
+    from typing import Callable
 
 
 @dataclass
@@ -134,8 +135,15 @@ def getpending(self) -> Event:
         ...
 
     @abstractmethod
-    def wait(self) -> None:
-        """Wait for an event."""
+    def wait(self, timeout: float | None) -> bool:
+        """Wait for an event. The return value is True if an event is
+        available, False if the timeout has been reached. If timeout is
+        None, wait forever. The timeout is in milliseconds."""
+        ...
+
+    @property
+    def input_hook(self) -> Callable[[], int] | None:
+        """Returns the current input hook."""
         ...
 
     @abstractmethod

Lib/_pyrepl/reader.py

@@ -650,7 +650,15 @@ def handle1(self, block: bool = True) -> bool:
             self.dirty = True
 
         while True:
-            event = self.console.get_event(block)
+            input_hook = self.console.input_hook
+            if input_hook:
+                input_hook()
+                # We use the same timeout as in readline.c: 100ms
+                while not self.console.wait(100):
+                    input_hook()
+                event = self.console.get_event(block=False)
+            else:
+                event = self.console.get_event(block)
             if not event:  # can only happen if we're not blocking
                 return False
 

Lib/_pyrepl/unix_console.py

@@ -118,9 +118,12 @@ def __init__(self):
 
         def register(self, fd, flag):
             self.fd = fd
-
-        def poll(self):  # note: a 'timeout' argument would be *milliseconds*
-            r, w, e = select.select([self.fd], [], [])
+        # note: The 'timeout' argument is received as *milliseconds*
+        def poll(self, timeout: float | None = None) -> list[int]:
+            if timeout is None:
+                r, w, e = select.select([self.fd], [], [])
+            else:
+                r, w, e = select.select([self.fd], [], [], timeout/1000)
             return r
 
     poll = MinimalPoll  # type: ignore[assignment]
@@ -385,11 +388,11 @@ def get_event(self, block: bool = True) -> Event | None:
                 break
         return self.event_queue.get()
 
-    def wait(self):
+    def wait(self, timeout: float | None = None) -> bool:
         """
         Wait for events on the console.
         """
-        self.pollob.poll()
+        return bool(self.pollob.poll(timeout))
 
     def set_cursor_vis(self, visible):
         """
@@ -527,6 +530,15 @@ def clear(self):
         self.__posxy = 0, 0
         self.screen = []
 
+    @property
+    def input_hook(self):
+        try:
+            import posix
+        except ImportError:
+            return None
+        if posix._is_inputhook_installed():
+            return posix._inputhook
+
     def __enable_bracketed_paste(self) -> None:
         os.write(self.output_fd, b"\x1b[?2004h")
 

Lib/_pyrepl/windows_console.py

@@ -23,6 +23,8 @@
 from multiprocessing import Value
 import os
 import sys
+import time
+import msvcrt
 
 from abc import ABC, abstractmethod
 from collections import deque
@@ -202,6 +204,15 @@ def refresh(self, screen: list[str], c_xy: tuple[int, int]) -> None:
         self.screen = screen
         self.move_cursor(cx, cy)
 
+    @property
+    def input_hook(self):
+        try:
+            import nt
+        except ImportError:
+            return None
+        if nt._is_inputhook_installed():
+            return nt._inputhook
+
     def __write_changed_line(
         self, y: int, oldline: str, newline: str, px_coord: int
     ) -> None:
@@ -460,9 +471,16 @@ def getpending(self) -> Event:
         processed."""
         return Event("key", "", b"")
 
-    def wait(self) -> None:
+    def wait(self, timeout: float | None) -> bool:
         """Wait for an event."""
-        raise NotImplementedError("No wait support")
+        # Poor man's Windows select loop
+        start_time = time.time()
+        while True:
+            if msvcrt.kbhit(): # type: ignore[attr-defined]
+                return True
+            if timeout and time.time() - start_time > timeout:
+                return False
+            time.sleep(0.01)
 
     def repaint(self) -> None:
         raise NotImplementedError("No repaint support")

Lib/test/test_pyrepl/test_reader.py

@@ -2,8 +2,10 @@
 import functools
 import rlcompleter
 from unittest import TestCase
+from unittest.mock import MagicMock, patch
 
 from .support import handle_all_events, handle_events_narrow_console, code_to_events, prepare_reader
+from test.support import import_helper
 from _pyrepl.console import Event
 from _pyrepl.reader import Reader
 
@@ -179,6 +181,21 @@ def test_newline_within_block_trailing_whitespace(self):
         self.assert_screen_equals(reader, expected)
         self.assertTrue(reader.finished)
 
+    def test_input_hook_is_called_if_set(self):
+        input_hook = MagicMock()
+        def _prepare_console(events):
+            console = MagicMock()
+            console.get_event.side_effect = events
+            console.height = 100
+            console.width = 80
+            console.input_hook = input_hook
+            return console
+
+        events = code_to_events("a")
+        reader, _ = handle_all_events(events, prepare_console=_prepare_console)
+
+        self.assertEqual(len(input_hook.mock_calls), 4)
+
     def test_keyboard_interrupt_clears_screen(self):
         namespace = {"itertools": itertools}
         code = "import itertools\nitertools."

Misc/NEWS.d/next/Core and Builtins/2024-05-31-12-06-11.gh-issue-119842.tCGVsv.rst

@@ -0,0 +1 @@
+Honor :c:func:`PyOS_InputHook` in the new REPL. Patch by Pablo Galindo

Modules/posixmodule.c

@@ -16784,6 +16784,37 @@ os__supports_virtual_terminal_impl(PyObject *module)
 }
 #endif
 
+/*[clinic input]
+os._inputhook
+
+Calls PyOS_CallInputHook droppong the GIL first
+[clinic start generated code]*/
+
+static PyObject *
+os__inputhook_impl(PyObject *module)
+/*[clinic end generated code: output=525aca4ef3c6149f input=fc531701930d064f]*/
+{
+     int result = 0;
+     if (PyOS_InputHook) {
+         Py_BEGIN_ALLOW_THREADS;
+         result = PyOS_InputHook();
+         Py_END_ALLOW_THREADS;
+     }
+     return PyLong_FromLong(result);
+}
+
+/*[clinic input]
+os._is_inputhook_installed
+
+Checks if PyOS_CallInputHook is set
+[clinic start generated code]*/
+
+static PyObject *
+os__is_inputhook_installed_impl(PyObject *module)
+/*[clinic end generated code: output=3b3eab4f672c689a input=ff177c9938dd76d8]*/
+{
+    return PyBool_FromLong(PyOS_InputHook != NULL);
+}
 
 static PyMethodDef posix_methods[] = {
 
@@ -16997,6 +17028,8 @@ static PyMethodDef posix_methods[] = {
     OS__PATH_LEXISTS_METHODDEF
 
     OS__SUPPORTS_VIRTUAL_TERMINAL_METHODDEF
+    OS__INPUTHOOK_METHODDEF
+    OS__IS_INPUTHOOK_INSTALLED_METHODDEF
     {NULL,              NULL}            /* Sentinel */
 };