diff options
Diffstat (limited to 'Lib/_pyrepl')
| -rw-r--r-- | Lib/_pyrepl/_module_completer.py | 24 | ||||
| -rw-r--r-- | Lib/_pyrepl/base_eventqueue.py | 18 | ||||
| -rw-r--r-- | Lib/_pyrepl/commands.py | 10 | ||||
| -rw-r--r-- | Lib/_pyrepl/main.py | 11 | ||||
| -rw-r--r-- | Lib/_pyrepl/reader.py | 9 | ||||
| -rw-r--r-- | Lib/_pyrepl/readline.py | 6 | ||||
| -rw-r--r-- | Lib/_pyrepl/simple_interact.py | 16 | ||||
| -rw-r--r-- | Lib/_pyrepl/unix_console.py | 21 | ||||
| -rw-r--r-- | Lib/_pyrepl/utils.py | 49 | ||||
| -rw-r--r-- | Lib/_pyrepl/windows_console.py | 67 |
10 files changed, 149 insertions, 82 deletions
diff --git a/Lib/_pyrepl/_module_completer.py b/Lib/_pyrepl/_module_completer.py index 347f05607c7..1e9462a4215 100644 --- a/Lib/_pyrepl/_module_completer.py +++ b/Lib/_pyrepl/_module_completer.py @@ -17,8 +17,8 @@ if TYPE_CHECKING: def make_default_module_completer() -> ModuleCompleter: - # Inside pyrepl, __package__ is set to '_pyrepl' - return ModuleCompleter(namespace={'__package__': '_pyrepl'}) + # Inside pyrepl, __package__ is set to None by default + return ModuleCompleter(namespace={'__package__': None}) class ModuleCompleter: @@ -42,11 +42,11 @@ class ModuleCompleter: self._global_cache: list[pkgutil.ModuleInfo] = [] self._curr_sys_path: list[str] = sys.path[:] - def get_completions(self, line: str) -> list[str]: + def get_completions(self, line: str) -> list[str] | None: """Return the next possible import completions for 'line'.""" result = ImportParser(line).parse() if not result: - return [] + return None try: return self.complete(*result) except Exception: @@ -81,8 +81,11 @@ class ModuleCompleter: def _find_modules(self, path: str, prefix: str) -> list[str]: if not path: # Top-level import (e.g. `import foo<tab>`` or `from foo<tab>`)` - return [name for _, name, _ in self.global_cache - if name.startswith(prefix)] + builtin_modules = [name for name in sys.builtin_module_names + if self.is_suggestion_match(name, prefix)] + third_party_modules = [module.name for module in self.global_cache + if self.is_suggestion_match(module.name, prefix)] + return sorted(builtin_modules + third_party_modules) if path.startswith('.'): # Convert relative path to absolute path @@ -97,7 +100,14 @@ class ModuleCompleter: if mod_info.ispkg and mod_info.name == segment] modules = self.iter_submodules(modules) return [module.name for module in modules - if module.name.startswith(prefix)] + if self.is_suggestion_match(module.name, prefix)] + + def is_suggestion_match(self, module_name: str, prefix: str) -> bool: + if prefix: + return module_name.startswith(prefix) + # For consistency with attribute completion, which + # does not suggest private attributes unless requested. + return not module_name.startswith("_") def iter_submodules(self, parent_modules: list[pkgutil.ModuleInfo]) -> Iterator[pkgutil.ModuleInfo]: """Iterate over all submodules of the given parent modules.""" diff --git a/Lib/_pyrepl/base_eventqueue.py b/Lib/_pyrepl/base_eventqueue.py index e018c4fc183..0589a0f437e 100644 --- a/Lib/_pyrepl/base_eventqueue.py +++ b/Lib/_pyrepl/base_eventqueue.py @@ -69,18 +69,14 @@ class BaseEventQueue: trace('added event {event}', event=event) self.events.append(event) - def push(self, char: int | bytes | str) -> None: + def push(self, char: int | bytes) -> None: """ Processes a character by updating the buffer and handling special key mappings. """ + assert isinstance(char, (int, bytes)) ord_char = char if isinstance(char, int) else ord(char) - if ord_char > 255: - assert isinstance(char, str) - char = bytes(char.encode(self.encoding, "replace")) - self.buf.extend(char) - else: - char = bytes(bytearray((ord_char,))) - self.buf.append(ord_char) + char = ord_char.to_bytes() + self.buf.append(ord_char) if char in self.keymap: if self.keymap is self.compiled_keymap: @@ -91,7 +87,7 @@ class BaseEventQueue: if isinstance(k, dict): self.keymap = k else: - self.insert(Event('key', k, self.flush_buf())) + self.insert(Event('key', k, bytes(self.flush_buf()))) self.keymap = self.compiled_keymap elif self.buf and self.buf[0] == 27: # escape @@ -100,7 +96,7 @@ class BaseEventQueue: # the docstring in keymap.py trace('unrecognized escape sequence, propagating...') self.keymap = self.compiled_keymap - self.insert(Event('key', '\033', bytearray(b'\033'))) + self.insert(Event('key', '\033', b'\033')) for _c in self.flush_buf()[1:]: self.push(_c) @@ -110,5 +106,5 @@ class BaseEventQueue: except UnicodeError: return else: - self.insert(Event('key', decoded, self.flush_buf())) + self.insert(Event('key', decoded, bytes(self.flush_buf()))) self.keymap = self.compiled_keymap diff --git a/Lib/_pyrepl/commands.py b/Lib/_pyrepl/commands.py index 2054a8e400f..50c824995d8 100644 --- a/Lib/_pyrepl/commands.py +++ b/Lib/_pyrepl/commands.py @@ -370,6 +370,13 @@ class self_insert(EditCommand): r = self.reader text = self.event * r.get_arg() r.insert(text) + if r.paste_mode: + data = "" + ev = r.console.getpending() + data += ev.data + if data: + r.insert(data) + r.last_refresh_cache.invalidated = True class insert_nl(EditCommand): @@ -439,7 +446,7 @@ class help(Command): import _sitebuiltins with self.reader.suspend(): - self.reader.msg = _sitebuiltins._Helper()() # type: ignore[assignment, call-arg] + self.reader.msg = _sitebuiltins._Helper()() # type: ignore[assignment] class invalid_key(Command): @@ -484,7 +491,6 @@ class perform_bracketed_paste(Command): data = "" start = time.time() while done not in data: - self.reader.console.wait(100) ev = self.reader.console.getpending() data += ev.data trace( diff --git a/Lib/_pyrepl/main.py b/Lib/_pyrepl/main.py index a6f824dcc4a..447eb1e551e 100644 --- a/Lib/_pyrepl/main.py +++ b/Lib/_pyrepl/main.py @@ -1,6 +1,7 @@ import errno import os import sys +import types CAN_USE_PYREPL: bool @@ -29,12 +30,10 @@ def interactive_console(mainmodule=None, quiet=False, pythonstartup=False): print(FAIL_REASON, file=sys.stderr) return sys._baserepl() - if mainmodule: - namespace = mainmodule.__dict__ - else: - import __main__ - namespace = __main__.__dict__ - namespace.pop("__pyrepl_interactive_console", None) + if not mainmodule: + mainmodule = types.ModuleType("__main__") + + namespace = mainmodule.__dict__ # sys._baserepl() above does this internally, we do it here startup_path = os.getenv("PYTHONSTARTUP") diff --git a/Lib/_pyrepl/reader.py b/Lib/_pyrepl/reader.py index 65c2230dfd6..0ebd9162eca 100644 --- a/Lib/_pyrepl/reader.py +++ b/Lib/_pyrepl/reader.py @@ -28,7 +28,7 @@ from contextlib import contextmanager from dataclasses import dataclass, field, fields from . import commands, console, input -from .utils import wlen, unbracket, disp_str, gen_colors +from .utils import wlen, unbracket, disp_str, gen_colors, THEME from .trace import trace @@ -491,11 +491,8 @@ class Reader: prompt = self.ps1 if self.can_colorize: - prompt = ( - f"{_colorize.theme["PROMPT"]}" - f"{prompt}" - f"{_colorize.theme["RESET"]}" - ) + t = THEME() + prompt = f"{t.prompt}{prompt}{t.reset}" return prompt def push_input_trans(self, itrans: input.KeymapTranslator) -> None: diff --git a/Lib/_pyrepl/readline.py b/Lib/_pyrepl/readline.py index 560a9db1921..9560ae779ab 100644 --- a/Lib/_pyrepl/readline.py +++ b/Lib/_pyrepl/readline.py @@ -134,7 +134,8 @@ class ReadlineAlikeReader(historical_reader.HistoricalReader, CompletingReader): return "".join(b[p + 1 : self.pos]) def get_completions(self, stem: str) -> list[str]: - if module_completions := self.get_module_completions(): + module_completions = self.get_module_completions() + if module_completions is not None: return module_completions if len(stem) == 0 and self.more_lines is not None: b = self.buffer @@ -165,7 +166,7 @@ class ReadlineAlikeReader(historical_reader.HistoricalReader, CompletingReader): result.sort() return result - def get_module_completions(self) -> list[str]: + def get_module_completions(self) -> list[str] | None: line = self.get_line() return self.config.module_completer.get_completions(line) @@ -606,6 +607,7 @@ def _setup(namespace: Mapping[str, Any]) -> None: # set up namespace in rlcompleter, which requires it to be a bona fide dict if not isinstance(namespace, dict): namespace = dict(namespace) + _wrapper.config.module_completer = ModuleCompleter(namespace) _wrapper.config.readline_completer = RLCompleter(namespace).complete # this is not really what readline.c does. Better than nothing I guess diff --git a/Lib/_pyrepl/simple_interact.py b/Lib/_pyrepl/simple_interact.py index e2274629b65..965b853c34b 100644 --- a/Lib/_pyrepl/simple_interact.py +++ b/Lib/_pyrepl/simple_interact.py @@ -31,6 +31,7 @@ import os import sys import code import warnings +import errno from .readline import _get_reader, multiline_input, append_history_file @@ -110,6 +111,10 @@ def run_multiline_interactive_console( more_lines = functools.partial(_more_lines, console) input_n = 0 + _is_x_showrefcount_set = sys._xoptions.get("showrefcount") + _is_pydebug_build = hasattr(sys, "gettotalrefcount") + show_ref_count = _is_x_showrefcount_set and _is_pydebug_build + def maybe_run_command(statement: str) -> bool: statement = statement.strip() if statement in console.locals or statement not in REPL_COMMANDS: @@ -149,6 +154,7 @@ def run_multiline_interactive_console( append_history_file() except (FileNotFoundError, PermissionError, OSError) as e: warnings.warn(f"failed to open the history file for writing: {e}") + input_n += 1 except KeyboardInterrupt: r = _get_reader() @@ -162,3 +168,13 @@ def run_multiline_interactive_console( except MemoryError: console.write("\nMemoryError\n") console.resetbuffer() + except SystemExit: + raise + except: + console.showtraceback() + console.resetbuffer() + if show_ref_count: + console.write( + f"[{sys.gettotalrefcount()} refs," + f" {sys.getallocatedblocks()} blocks]\n" + ) diff --git a/Lib/_pyrepl/unix_console.py b/Lib/_pyrepl/unix_console.py index 07b160d2324..d21cdd9b076 100644 --- a/Lib/_pyrepl/unix_console.py +++ b/Lib/_pyrepl/unix_console.py @@ -29,6 +29,7 @@ import signal import struct import termios import time +import types import platform from fcntl import ioctl @@ -39,6 +40,12 @@ from .trace import trace from .unix_eventqueue import EventQueue from .utils import wlen +# declare posix optional to allow None assignment on other platforms +posix: types.ModuleType | None +try: + import posix +except ImportError: + posix = None TYPE_CHECKING = False @@ -197,6 +204,12 @@ class UnixConsole(Console): self.event_queue = EventQueue(self.input_fd, self.encoding) self.cursor_visible = 1 + signal.signal(signal.SIGCONT, self._sigcont_handler) + + def _sigcont_handler(self, signum, frame): + self.restore() + self.prepare() + def __read(self, n: int) -> bytes: return os.read(self.input_fd, n) @@ -550,11 +563,9 @@ class UnixConsole(Console): @property def input_hook(self): - try: - import posix - except ImportError: - return None - if posix._is_inputhook_installed(): + # avoid inline imports here so the repl doesn't get flooded + # with import logging from -X importtime=2 + if posix is not None and posix._is_inputhook_installed(): return posix._inputhook def __enable_bracketed_paste(self) -> None: diff --git a/Lib/_pyrepl/utils.py b/Lib/_pyrepl/utils.py index fe154aa59a0..e04fbdc6c8a 100644 --- a/Lib/_pyrepl/utils.py +++ b/Lib/_pyrepl/utils.py @@ -23,6 +23,11 @@ IDENTIFIERS_AFTER = {"def", "class"} BUILTINS = {str(name) for name in dir(builtins) if not name.startswith('_')} +def THEME(**kwargs): + # Not cached: the user can modify the theme inside the interactive session. + return _colorize.get_theme(**kwargs).syntax + + class Span(NamedTuple): """Span indexing that's inclusive on both ends.""" @@ -36,15 +41,21 @@ class Span(NamedTuple): @classmethod def from_token(cls, token: TI, line_len: list[int]) -> Self: + end_offset = -1 + if (token.type in {T.FSTRING_MIDDLE, T.TSTRING_MIDDLE} + and token.string.endswith(("{", "}"))): + # gh-134158: a visible trailing brace comes from a double brace in input + end_offset += 1 + return cls( line_len[token.start[0] - 1] + token.start[1], - line_len[token.end[0] - 1] + token.end[1] - 1, + line_len[token.end[0] - 1] + token.end[1] + end_offset, ) class ColorSpan(NamedTuple): span: Span - tag: _colorize.ColorTag + tag: str @functools.cache @@ -97,6 +108,8 @@ def gen_colors(buffer: str) -> Iterator[ColorSpan]: for color in gen_colors_from_token_stream(gen, line_lengths): yield color last_emitted = color + except SyntaxError: + return except tokenize.TokenError as te: yield from recover_unterminated_string( te, line_lengths, last_emitted, buffer @@ -135,7 +148,7 @@ def recover_unterminated_string( span = Span(start, end) trace("yielding span {a} -> {b}", a=span.start, b=span.end) - yield ColorSpan(span, "STRING") + yield ColorSpan(span, "string") else: trace( "unhandled token error({buffer}) = {te}", @@ -164,28 +177,28 @@ def gen_colors_from_token_stream( | T.TSTRING_START | T.TSTRING_MIDDLE | T.TSTRING_END ): span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "STRING") + yield ColorSpan(span, "string") case T.COMMENT: span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "COMMENT") + yield ColorSpan(span, "comment") case T.NUMBER: span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "NUMBER") + yield ColorSpan(span, "number") case T.OP: if token.string in "([{": bracket_level += 1 elif token.string in ")]}": bracket_level -= 1 span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "OP") + yield ColorSpan(span, "op") case T.NAME: if is_def_name: is_def_name = False span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "DEFINITION") + yield ColorSpan(span, "definition") elif keyword.iskeyword(token.string): span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "KEYWORD") + yield ColorSpan(span, "keyword") if token.string in IDENTIFIERS_AFTER: is_def_name = True elif ( @@ -194,10 +207,10 @@ def gen_colors_from_token_stream( and is_soft_keyword_used(prev_token, token, next_token) ): span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "SOFT_KEYWORD") + yield ColorSpan(span, "soft_keyword") elif token.string in BUILTINS: span = Span.from_token(token, line_lengths) - yield ColorSpan(span, "BUILTIN") + yield ColorSpan(span, "builtin") keyword_first_sets_match = {"False", "None", "True", "await", "lambda", "not"} @@ -249,7 +262,10 @@ def is_soft_keyword_used(*tokens: TI | None) -> bool: def disp_str( - buffer: str, colors: list[ColorSpan] | None = None, start_index: int = 0 + buffer: str, + colors: list[ColorSpan] | None = None, + start_index: int = 0, + force_color: bool = False, ) -> tuple[CharBuffer, CharWidths]: r"""Decompose the input buffer into a printable variant with applied colors. @@ -290,15 +306,16 @@ def disp_str( # move past irrelevant spans colors.pop(0) + theme = THEME(force_color=force_color) pre_color = "" post_color = "" if colors and colors[0].span.start < start_index: # looks like we're continuing a previous color (e.g. a multiline str) - pre_color = _colorize.theme[colors[0].tag] + pre_color = theme[colors[0].tag] for i, c in enumerate(buffer, start_index): if colors and colors[0].span.start == i: # new color starts now - pre_color = _colorize.theme[colors[0].tag] + pre_color = theme[colors[0].tag] if c == "\x1a": # CTRL-Z on Windows chars.append(c) @@ -315,7 +332,7 @@ def disp_str( char_widths.append(str_width(c)) if colors and colors[0].span.end == i: # current color ends now - post_color = _colorize.theme["RESET"] + post_color = theme.reset colors.pop(0) chars[-1] = pre_color + chars[-1] + post_color @@ -325,7 +342,7 @@ def disp_str( if colors and colors[0].span.start < i and colors[0].span.end > i: # even though the current color should be continued, reset it for now. # the next call to `disp_str()` will revive it. - chars[-1] += _colorize.theme["RESET"] + chars[-1] += theme.reset return chars, char_widths diff --git a/Lib/_pyrepl/windows_console.py b/Lib/_pyrepl/windows_console.py index 77985e59a93..c56dcd6d7dd 100644 --- a/Lib/_pyrepl/windows_console.py +++ b/Lib/_pyrepl/windows_console.py @@ -24,6 +24,7 @@ import os import sys import ctypes +import types from ctypes.wintypes import ( _COORD, WORD, @@ -58,6 +59,12 @@ except: self.err = err self.descr = descr +# declare nt optional to allow None assignment on other platforms +nt: types.ModuleType | None +try: + import nt +except ImportError: + nt = None TYPE_CHECKING = False @@ -121,9 +128,8 @@ class _error(Exception): def _supports_vt(): try: - import nt return nt._supports_virtual_terminal() - except (ImportError, AttributeError): + except AttributeError: return False class WindowsConsole(Console): @@ -235,11 +241,9 @@ class WindowsConsole(Console): @property def input_hook(self): - try: - import nt - except ImportError: - return None - if nt._is_inputhook_installed(): + # avoid inline imports here so the repl doesn't get flooded + # with import logging from -X importtime=2 + if nt is not None and nt._is_inputhook_installed(): return nt._inputhook def __write_changed_line( @@ -415,10 +419,7 @@ class WindowsConsole(Console): return info.srWindow.Bottom # type: ignore[no-any-return] - def _read_input(self, block: bool = True) -> INPUT_RECORD | None: - if not block and not self.wait(timeout=0): - return None - + def _read_input(self) -> INPUT_RECORD | None: rec = INPUT_RECORD() read = DWORD() if not ReadConsoleInput(InHandle, rec, 1, read): @@ -427,14 +428,10 @@ class WindowsConsole(Console): return rec def _read_input_bulk( - self, block: bool, n: int + self, n: int ) -> tuple[ctypes.Array[INPUT_RECORD], int]: rec = (n * INPUT_RECORD)() read = DWORD() - - if not block and not self.wait(timeout=0): - return rec, 0 - if not ReadConsoleInput(InHandle, rec, n, read): raise WinError(GetLastError()) @@ -445,8 +442,11 @@ class WindowsConsole(Console): and there is no event pending, otherwise waits for the completion of an event.""" + if not block and not self.wait(timeout=0): + return None + while self.event_queue.empty(): - rec = self._read_input(block) + rec = self._read_input() if rec is None: return None @@ -464,7 +464,7 @@ class WindowsConsole(Console): if key == "\r": # Make enter unix-like - return Event(evt="key", data="\n", raw=b"\n") + return Event(evt="key", data="\n") elif key_event.wVirtualKeyCode == 8: # Turn backspace directly into the command key = "backspace" @@ -476,24 +476,29 @@ class WindowsConsole(Console): key = f"ctrl {key}" elif key_event.dwControlKeyState & ALT_ACTIVE: # queue the key, return the meta command - self.event_queue.insert(Event(evt="key", data=key, raw=key)) + self.event_queue.insert(Event(evt="key", data=key)) return Event(evt="key", data="\033") # keymap.py uses this for meta - return Event(evt="key", data=key, raw=key) + return Event(evt="key", data=key) if block: continue return None elif self.__vt_support: # If virtual terminal is enabled, scanning VT sequences - self.event_queue.push(rec.Event.KeyEvent.uChar.UnicodeChar) + for char in raw_key.encode(self.event_queue.encoding, "replace"): + self.event_queue.push(char) continue if key_event.dwControlKeyState & ALT_ACTIVE: - # queue the key, return the meta command - self.event_queue.insert(Event(evt="key", data=key, raw=raw_key)) - return Event(evt="key", data="\033") # keymap.py uses this for meta - - return Event(evt="key", data=key, raw=raw_key) + # Do not swallow characters that have been entered via AltGr: + # Windows internally converts AltGr to CTRL+ALT, see + # https://learn.microsoft.com/en-us/windows/win32/api/winuser/nf-winuser-vkkeyscanw + if not key_event.dwControlKeyState & CTRL_ACTIVE: + # queue the key, return the meta command + self.event_queue.insert(Event(evt="key", data=key)) + return Event(evt="key", data="\033") # keymap.py uses this for meta + + return Event(evt="key", data=key) return self.event_queue.get() def push_char(self, char: int | bytes) -> None: @@ -542,12 +547,20 @@ class WindowsConsole(Console): if e2: e.data += e2.data - recs, rec_count = self._read_input_bulk(False, 1024) + recs, rec_count = self._read_input_bulk(1024) for i in range(rec_count): rec = recs[i] + # In case of a legacy console, we do not only receive a keydown + # event, but also a keyup event - and for uppercase letters + # an additional SHIFT_PRESSED event. if rec and rec.EventType == KEY_EVENT: key_event = rec.Event.KeyEvent + if not key_event.bKeyDown: + continue ch = key_event.uChar.UnicodeChar + if ch == "\x00": + # ignore SHIFT_PRESSED and special keys + continue if ch == "\r": ch += "\n" e.data += ch |