gh-118894: Make asyncio REPL use pyrepl

Lib/_pyrepl/console.py

@@ -19,7 +19,10 @@
 
 from __future__ import annotations
 
+import _colorize  # type: ignore[import-not-found]
 from abc import ABC, abstractmethod
+import ast
+import code
 from dataclasses import dataclass, field
 
 
@@ -110,3 +113,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)
+            except (OverflowError, ValueError):
+                    self.showsyntaxerror(filename)
+                    return False
+
+            if code is None:
+                return True
+
+            self.runcode(code)
+        return False

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
 from .unix_console import _error
 
@@ -65,57 +64,20 @@ def _strip_final_indent(text: str) -> str:
     "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)
-            except (OverflowError, ValueError):
-                    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,21 +1,23 @@
 import ast
 import asyncio
-import code
 import concurrent.futures
 import inspect
+import os
 import site
 import sys
 import threading
 import types
 import warnings
 
+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
@@ -75,12 +77,36 @@ def run(self):
                 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'
             )
+            exit_message = '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)
 
-            console.interact(
-                banner=banner,
-                exitmsg='exiting asyncio REPL...')
+            try:
+                import errno
+                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 BaseException:
+                    console.showtraceback()
+                console.write(exit_message + '\n')
         finally:
             warnings.filterwarnings(
                 'ignore',
@@ -126,7 +152,7 @@ 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()
 

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.