gh-118894: Make asyncio REPL use pyrepl

Lib/_pyrepl/commands.py

@@ -219,6 +219,11 @@ def do(self) -> None:
         os.kill(os.getpid(), signal.SIGINT)
 
 
+class ctrl_c(Command):
+    def do(self) -> None:
+        raise KeyboardInterrupt
+
+
 class suspend(Command):
     def do(self) -> None:
         import signal

Lib/_pyrepl/console.py

@@ -19,10 +19,13 @@
 
 from __future__ import annotations
 
-import sys
+import _colorize  # type: ignore[import-not-found]
 
 from abc import ABC, abstractmethod
+import ast
+import code
 from dataclasses import dataclass, field
+import sys
 
 
 TYPE_CHECKING = False
@@ -136,3 +139,45 @@ def wait(self) -> None:
 
     @abstractmethod
     def repaint(self) -> None: ...
+
+
+class InteractiveColoredConsole(code.InteractiveConsole):
+    def __init__(
+        self,
+        locals: dict[str, object] | None = None,
+        filename: str = "<console>",
+        *,
+        local_exit: bool = False,
+    ) -> None:
+        super().__init__(locals=locals, filename=filename, local_exit=local_exit)  # type: ignore[call-arg]
+        self.can_colorize = _colorize.can_colorize()
+
+    def showsyntaxerror(self, filename=None):
+        super().showsyntaxerror(colorize=self.can_colorize)
+
+    def showtraceback(self):
+        super().showtraceback(colorize=self.can_colorize)
+
+    def runsource(self, source, filename="<input>", symbol="single"):
+        try:
+            tree = ast.parse(source)
+        except (OverflowError, SyntaxError, ValueError):
+            self.showsyntaxerror(filename)
+            return False
+        if tree.body:
+            *_, last_stmt = tree.body
+        for stmt in tree.body:
+            wrapper = ast.Interactive if stmt is last_stmt else ast.Module
+            the_symbol = symbol if stmt is last_stmt else "exec"
+            item = wrapper([stmt])
+            try:
+                code = self.compile.compiler(item, filename, the_symbol, dont_inherit=True)
+            except (OverflowError, ValueError, SyntaxError):
+                    self.showsyntaxerror(filename)
+                    return False
+
+            if code is None:
+                return True
+
+            self.runcode(code)
+        return False

Lib/_pyrepl/reader.py

@@ -131,6 +131,7 @@ def make_default_commands() -> dict[CommandName, type[Command]]:
         ("\\\\", "self-insert"),
         (r"\x1b[200~", "enable_bracketed_paste"),
         (r"\x1b[201~", "disable_bracketed_paste"),
+        (r"\x03", "ctrl-c"),
     ]
     + [(c, "self-insert") for c in map(chr, range(32, 127)) if c != "\\"]
     + [(c, "self-insert") for c in map(chr, range(128, 256)) if c.isalpha()]

Lib/_pyrepl/simple_interact.py

@@ -25,14 +25,13 @@
 
 from __future__ import annotations
 
-import _colorize  # type: ignore[import-not-found]
 import _sitebuiltins
 import linecache
 import sys
 import code
-import ast
 from types import ModuleType
 
+from .console import InteractiveColoredConsole
 from .readline import _get_reader, multiline_input
 
 _error: tuple[type[Exception], ...] | type[Exception]
@@ -74,57 +73,21 @@ def _clear_screen():
     "clear": _clear_screen,
 }
 
-class InteractiveColoredConsole(code.InteractiveConsole):
-    def __init__(
-        self,
-        locals: dict[str, object] | None = None,
-        filename: str = "<console>",
-        *,
-        local_exit: bool = False,
-    ) -> None:
-        super().__init__(locals=locals, filename=filename, local_exit=local_exit)  # type: ignore[call-arg]
-        self.can_colorize = _colorize.can_colorize()
-
-    def showsyntaxerror(self, filename=None):
-        super().showsyntaxerror(colorize=self.can_colorize)
-
-    def showtraceback(self):
-        super().showtraceback(colorize=self.can_colorize)
-
-    def runsource(self, source, filename="<input>", symbol="single"):
-        try:
-            tree = ast.parse(source)
-        except (OverflowError, SyntaxError, ValueError):
-            self.showsyntaxerror(filename)
-            return False
-        if tree.body:
-            *_, last_stmt = tree.body
-        for stmt in tree.body:
-            wrapper = ast.Interactive if stmt is last_stmt else ast.Module
-            the_symbol = symbol if stmt is last_stmt else "exec"
-            item = wrapper([stmt])
-            try:
-                code = compile(item, filename, the_symbol, dont_inherit=True)
-            except (OverflowError, ValueError, SyntaxError):
-                    self.showsyntaxerror(filename)
-                    return False
-
-            if code is None:
-                return True
-
-            self.runcode(code)
-        return False
-
 
 def run_multiline_interactive_console(
-    mainmodule: ModuleType | None= None, future_flags: int = 0
+    mainmodule: ModuleType | None = None,
+    future_flags: int = 0,
+    console: code.InteractiveConsole | None = None,
 ) -> None:
     import __main__
     from .readline import _setup
     _setup()
 
     mainmodule = mainmodule or __main__
-    console = InteractiveColoredConsole(mainmodule.__dict__, filename="<stdin>")
+    if console is None:
+        console = InteractiveColoredConsole(
+            mainmodule.__dict__, filename="<stdin>"
+        )
     if future_flags:
         console.compile.compiler.flags |= future_flags
 

Lib/asyncio/__main__.py

@@ -1,42 +1,49 @@
 import ast
 import asyncio
-import code
 import concurrent.futures
 import inspect
+import os
 import site
 import sys
 import threading
 import types
 import warnings
 
+from _colorize import can_colorize, ANSIColors  # type: ignore[import-not-found]
+from _pyrepl.console import InteractiveColoredConsole
+
 from . import futures
 
 
-class AsyncIOInteractiveConsole(code.InteractiveConsole):
+class AsyncIOInteractiveConsole(InteractiveColoredConsole):
 
     def __init__(self, locals, loop):
-        super().__init__(locals)
+        super().__init__(locals, filename="<stdin>")
         self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
 
         self.loop = loop
 
     def runcode(self, code):
+        global return_code
         future = concurrent.futures.Future()
 
         def callback():
+            global return_code
             global repl_future
-            global repl_future_interrupted
+            global keyboard_interrupted
 
             repl_future = None
-            repl_future_interrupted = False
+            keyboard_interrupted = False
 
             func = types.FunctionType(code, self.locals)
             try:
                 coro = func()
-            except SystemExit:
-                raise
+            except SystemExit as se:
+                return_code = se.code
+                self.loop.stop()
+                return
             except KeyboardInterrupt as ex:
-                repl_future_interrupted = True
+                keyboard_interrupted = True
                 future.set_exception(ex)
                 return
             except BaseException as ex:
@@ -57,10 +64,12 @@ def callback():
 
         try:
             return future.result()
-        except SystemExit:
-            raise
+        except SystemExit as se:
+            return_code = se.code
+            self.loop.stop()
+            return
         except BaseException:
-            if repl_future_interrupted:
+            if keyboard_interrupted:
                 self.write("\nKeyboardInterrupt\n")
             else:
                 self.showtraceback()
@@ -69,18 +78,56 @@ def callback():
 class REPLThread(threading.Thread):
 
     def run(self):
+        global return_code
+
         try:
             banner = (
                 f'asyncio REPL {sys.version} on {sys.platform}\n'
                 f'Use "await" directly instead of "asyncio.run()".\n'
                 f'Type "help", "copyright", "credits" or "license" '
                 f'for more information.\n'
-                f'{getattr(sys, "ps1", ">>> ")}import asyncio'
             )
 
-            console.interact(
-                banner=banner,
-                exitmsg='exiting asyncio REPL...')
+            console.write(banner)
+
+            if startup_path := os.getenv("PYTHONSTARTUP"):
+                import tokenize
+                with tokenize.open(startup_path) as f:
+                    startup_code = compile(f.read(), startup_path, "exec")
+                    exec(startup_code, console.locals)
+
+            ps1 = getattr(sys, "ps1", ">>> ")
+            if can_colorize():
+                ps1 = f"{ANSIColors.BOLD_MAGENTA}{ps1}{ANSIColors.RESET}"
+            console.write(f"{ps1}import asyncio\n")
+
+            try:
+                import errno
+                if os.getenv("PYTHON_BASIC_REPL"):
+                    raise RuntimeError("user environment requested basic REPL")
+                if not os.isatty(sys.stdin.fileno()):
+                    raise OSError(errno.ENOTTY, "tty required", "stdin")
+
+                # This import will fail on operating systems with no termios.
+                from _pyrepl.simple_interact import (
+                    check,
+                    run_multiline_interactive_console,
+                )
+                if err := check():
+                    raise RuntimeError(err)
+            except Exception as e:
+                console.interact(banner="", exitmsg=exit_message)
+            else:
+                try:
+                    run_multiline_interactive_console(console=console)
+                except SystemExit:
+                    # expected via the `exit` and `quit` commands
+                    pass
+                except BaseException:
+                    # unexpected issue
+                    console.showtraceback()
+                    console.write("Internal error, ")
+                    return_code = 1
         finally:
             warnings.filterwarnings(
                 'ignore',
@@ -91,6 +138,9 @@ def run(self):
 
 
 if __name__ == '__main__':
+    CAN_USE_PYREPL = True
+
+    return_code = 0
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
 
@@ -103,7 +153,7 @@ def run(self):
     console = AsyncIOInteractiveConsole(repl_locals, loop)
 
     repl_future = None
-    repl_future_interrupted = False
+    keyboard_interrupted = False
 
     try:
         import readline  # NoQA
@@ -126,17 +176,20 @@ def run(self):
                 completer = rlcompleter.Completer(console.locals)
                 readline.set_completer(completer.complete)
 
-    repl_thread = REPLThread()
+    repl_thread = REPLThread(name="Interactive thread")
     repl_thread.daemon = True
     repl_thread.start()
 
     while True:
         try:
             loop.run_forever()
         except KeyboardInterrupt:
+            keyboard_interrupted = True
             if repl_future and not repl_future.done():
                 repl_future.cancel()
-                repl_future_interrupted = True
             continue
         else:
             break
+
+    console.write('exiting asyncio REPL...\n')
+    sys.exit(return_code)

Lib/test/test_pyrepl/test_interact.py

@@ -6,7 +6,7 @@
 
 from test.support import force_not_colorized
 
-from _pyrepl.simple_interact import InteractiveColoredConsole
+from _pyrepl.console import InteractiveColoredConsole
 
 
 class TestSimpleInteract(unittest.TestCase):

Misc/NEWS.d/next/Library/2024-05-22-21-20-43.gh-issue-118894.xHdxR_.rst

@@ -0,0 +1 @@
+:mod:`asyncio` REPL now has the same capabilities as PyREPL.