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
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 os.path
+import sys
TYPE_CHECKING = False
@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 (SyntaxError, OverflowError, 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 SyntaxError as e:
+ if e.args[0] == "'await' outside function":
+ python = os.path.basename(sys.executable)
+ e.add_note(
+ f"Try the asyncio REPL ({python} -m asyncio) to use"
+ f" top-level 'await' and run background asyncio tasks."
+ )
+ self.showsyntaxerror(filename)
+ return False
+ except (OverflowError, ValueError):
+ self.showsyntaxerror(filename)
+ return False
+
+ if code is None:
+ return True
+
+ self.runcode(code)
+ return False
("\\\\", "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()]
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]
"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
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:
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()
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',
if __name__ == '__main__':
+ CAN_USE_PYREPL = True
+
+ return_code = 0
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
console = AsyncIOInteractiveConsole(repl_locals, loop)
repl_future = None
- repl_future_interrupted = False
+ keyboard_interrupted = False
try:
import readline # NoQA
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()
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)
from test.support import force_not_colorized
-from _pyrepl.simple_interact import InteractiveColoredConsole
+from _pyrepl.console import InteractiveColoredConsole
class TestSimpleInteract(unittest.TestCase):
--- /dev/null
+:mod:`asyncio` REPL now has the same capabilities as PyREPL.
projects / thirdparty / Python / cpython.git / commitdiff
