gh-122622: Add PyREPL simple_interact.statement_submitted_hook

Lib/_pyrepl/simple_interact.py

@@ -33,6 +33,25 @@
 from .readline import _get_reader, multiline_input, append_history_file
 
 
+# Called with the complete statement, which may span multiple lines, after the
+# user submits it but before it is executed. The hook is not called for PyREPL
+# commands such as ``clear``. Exceptions raised by the hook are displayed but
+# do not prevent the statement from running. Its return value is ignored.
+# External tools such as IDEs can install a hook to augment the behavior of the
+# REPL.
+#
+# For example, an IDE can mark the start of command execution:
+#
+#     from _pyrepl import simple_interact
+#
+#     simple_interact.statement_submitted_hook = lambda x: print("\x1b]633;C\x07", end="")
+#
+# In the REPL:
+#     >>> print("Hello, world!")
+#     \x1b]633;C\x07Hello, world!
+statement_submitted_hook = None
+
+
 _error: tuple[type[Exception], ...] | type[Exception]
 try:
     from .unix_console import _error
@@ -145,6 +164,12 @@ def maybe_run_command(statement: str) -> bool:
             if maybe_run_command(statement):
                 continue
 
+            if statement_submitted_hook is not None:
+                try:
+                    statement_submitted_hook(statement)
+                except Exception:
+                    console.showtraceback()
+
             input_name = f"<python-input-{input_n}>"
             more = console.push(_strip_final_indent(statement), filename=input_name, _symbol="single")  # type: ignore[call-arg]
             assert not more

Lib/test/test_pyrepl/test_interact.py

@@ -2,11 +2,12 @@
 import io
 import warnings
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 from textwrap import dedent
 
 from test.support import force_not_colorized
 
+from _pyrepl import simple_interact
 from _pyrepl.console import InteractiveColoredConsole
 from _pyrepl.simple_interact import _more_lines
 
@@ -299,3 +300,78 @@ def f():
         count = sum("'return' in a 'finally' block" in str(w.message)
                     for w in caught)
         self.assertEqual(count, 1)
+
+
+class TestStatementSubmittedHook(unittest.TestCase):
+    def _run_interactive(self, statements, hook):
+        console = InteractiveColoredConsole()
+        statement_iter = iter(statements)
+
+        def fake_multiline_input(more_lines, ps1, ps2):
+            try:
+                return next(statement_iter)
+            except StopIteration:
+                raise EOFError
+
+        output = io.StringIO()
+        with (
+            patch.object(simple_interact, "statement_submitted_hook", hook),
+            patch.object(
+                simple_interact,
+                "multiline_input",
+                side_effect=fake_multiline_input,
+            ),
+            patch.object(simple_interact, "_get_reader"),
+            patch.object(simple_interact, "append_history_file"),
+            patch("_pyrepl.readline._setup"),
+            contextlib.redirect_stdout(output),
+            contextlib.redirect_stderr(output),
+        ):
+            simple_interact.run_multiline_interactive_console(console)
+
+        return output.getvalue(), console.locals
+
+    def test_hook_called_before_statement_execution(self):
+        statement = "if True:\n    print('statement executed')\n"
+        escape_sequence = "\x1b]633;C\x07"
+        hook = MagicMock(
+            side_effect=lambda statement: print(escape_sequence, end="")
+        )
+        output, _ = self._run_interactive([statement], hook)
+
+        hook.assert_called_once_with(statement)
+        self.assertEqual(output, f"{escape_sequence}statement executed\n")
+
+    @force_not_colorized
+    def test_hook_exception_is_displayed(self):
+        hook = MagicMock(side_effect=RuntimeError("hook error"))
+
+        output, namespace = self._run_interactive(["x = 1"], hook)
+
+        self.assertIn("RuntimeError: hook error", output)
+        self.assertEqual(namespace["x"], 1)
+
+    def test_hook_return_value_is_ignored(self):
+        hook = MagicMock(return_value=True)
+
+        _, namespace = self._run_interactive(["x = 1"], hook)
+
+        self.assertEqual(namespace["x"], 1)
+
+    def test_hook_not_called_for_repl_commands(self):
+        hook = MagicMock()
+        self._run_interactive(["clear"], hook)
+
+        hook.assert_not_called()
+
+    @force_not_colorized
+    def test_hook_called_before_execution_exception(self):
+        statement = "1 / 0"
+        marker = "hook ran first"
+        hook = MagicMock(side_effect=lambda statement: print(marker))
+        output, _ = self._run_interactive([statement], hook)
+
+        hook.assert_called_once_with(statement)
+        marker_index = output.index(marker)
+        traceback_index = output.index("ZeroDivisionError")
+        self.assertLess(marker_index, traceback_index)

Misc/NEWS.d/next/Library/2026-09-09-23-15-03.gh-issue-122622.YfSSrl.rst

@@ -0,0 +1,3 @@
+Add an optional ``_pyrepl.simple_interact.statement_submitted_hook`` callback
+that runs after a complete statement is submitted and before it is executed,
+allowing external tools to implement features such as shell integration.