From f4ccaa674d8d45186b964e37bedc519f371498c0 Mon Sep 17 00:00:00 2001 From: M Bussonnier Date: Thu, 19 Feb 2026 10:40:18 +0100 Subject: [PATCH 1/2] Add type annotations to _process_common.py and _process_win32.py Note in particular in that in _process_common, we change the name of the fist argument of arg_split to be `commandline` instead of `s` for consistency --- IPython/core/magic.py | 2 +- IPython/utils/_process_common.py | 15 ++++++++------- IPython/utils/_process_win32.py | 7 ++++--- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/IPython/core/magic.py b/IPython/core/magic.py index 74c97c19003..5204eb592b2 100644 --- a/IPython/core/magic.py +++ b/IPython/core/magic.py @@ -736,7 +736,7 @@ def parse_options( if len(args) >= 1: # If the list of inputs only has 0 or 1 thing in it, there's no # need to look for options - argv = arg_split(arg_str, posix, strict) # type: ignore[no-untyped-call] + argv = arg_split(arg_str, posix, strict) # Do regular option processing try: opts, args = getopt(argv, opt_str, long_opts) diff --git a/IPython/utils/_process_common.py b/IPython/utils/_process_common.py index 77305853cf8..93e5573d091 100644 --- a/IPython/utils/_process_common.py +++ b/IPython/utils/_process_common.py @@ -27,7 +27,7 @@ # Function definitions #----------------------------------------------------------------------------- -def read_no_interrupt(stream: IO[Any]) -> bytes: +def read_no_interrupt(stream: IO[Any]) -> bytes | None: """Read from a pipe ignoring EINTR errors. This is necessary because when reading from pipes with GUI event loops @@ -45,8 +45,8 @@ def read_no_interrupt(stream: IO[Any]) -> bytes: def process_handler( cmd: Union[str, List[str]], callback: Callable[[subprocess.Popen], int | str | bytes], - stderr=subprocess.PIPE, -) -> int | str | bytes: + stderr: int = subprocess.PIPE, +) -> int | str | bytes | None: """Open a command in a shell subprocess and execute a callback. This function provides common scaffolding for creating subprocess.Popen() @@ -118,7 +118,7 @@ def process_handler( return out -def getoutput(cmd): +def getoutput(cmd: str | list[str]) -> str: """Run a command and return its stdout/stderr as a string. Parameters @@ -141,7 +141,7 @@ def getoutput(cmd): return py3compat.decode(out) -def getoutputerror(cmd): +def getoutputerror(cmd: str | list[str]) -> tuple[str, str]: """Return (standard output, standard error) of executing cmd in a shell. Accepts the same arguments as os.system(). @@ -158,7 +158,8 @@ def getoutputerror(cmd): """ return get_output_error_code(cmd)[:2] -def get_output_error_code(cmd): + +def get_output_error_code(cmd: str | list[str]) -> tuple[str, str, int | None]: """Return (standard output, standard error, return code) of executing cmd in a shell. @@ -182,7 +183,7 @@ def get_output_error_code(cmd): out, err = out_err return py3compat.decode(out), py3compat.decode(err), p.returncode -def arg_split(s, posix=False, strict=True): +def arg_split(commandline: str, posix: bool = False, strict: bool = True) -> list[str]: """Split a command line's arguments in a shell-like manner. This is a modified version of the standard library's shlex.split() diff --git a/IPython/utils/_process_win32.py b/IPython/utils/_process_win32.py index 0600b1ee14c..ad13cc7a210 100644 --- a/IPython/utils/_process_win32.py +++ b/IPython/utils/_process_win32.py @@ -59,7 +59,7 @@ def __exit__( self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], - traceback: TracebackType, + traceback: Optional[TracebackType], ) -> None: if self.is_unc_path: os.chdir(self.path) @@ -75,7 +75,7 @@ def _system_body(p: subprocess.Popen) -> int: def stdout_read() -> None: try: assert p.stdout is not None - for byte_line in read_no_interrupt(p.stdout).splitlines(): + for byte_line in (read_no_interrupt(p.stdout) or b"").splitlines(): line = byte_line.decode(enc, "replace") print(line, file=sys.stdout) except Exception as e: @@ -84,7 +84,7 @@ def stdout_read() -> None: def stderr_read() -> None: try: assert p.stderr is not None - for byte_line in read_no_interrupt(p.stderr).splitlines(): + for byte_line in (read_no_interrupt(p.stderr) or b"").splitlines(): line = byte_line.decode(enc, "replace") print(line, file=sys.stderr) except Exception as e: @@ -210,6 +210,7 @@ def arg_split( arg_split = py_arg_split + def check_pid(pid: int) -> bool: # OpenProcess returns 0 if no such process (of ours) exists # positive int otherwise From a1076b0178ac20655bb3c4bdadca4e963255a25b Mon Sep 17 00:00:00 2001 From: M Bussonnier Date: Thu, 19 Feb 2026 10:52:46 +0100 Subject: [PATCH 2/2] one more pass --- IPython/utils/_process_common.py | 22 ++++++++++++---------- IPython/utils/_process_win32.py | 4 +--- IPython/utils/py3compat.py | 2 +- pyproject.toml | 3 ++- 4 files changed, 16 insertions(+), 15 deletions(-) diff --git a/IPython/utils/_process_common.py b/IPython/utils/_process_common.py index 93e5573d091..fcbf26a2f0d 100644 --- a/IPython/utils/_process_common.py +++ b/IPython/utils/_process_common.py @@ -18,16 +18,18 @@ import shlex import subprocess import sys -from typing import IO, Any, List, Union +from typing import IO, List, TypeVar, Union from collections.abc import Callable +_T = TypeVar("_T") + from IPython.utils import py3compat #----------------------------------------------------------------------------- # Function definitions #----------------------------------------------------------------------------- -def read_no_interrupt(stream: IO[Any]) -> bytes | None: +def read_no_interrupt(stream: IO[bytes]) -> bytes | None: """Read from a pipe ignoring EINTR errors. This is necessary because when reading from pipes with GUI event loops @@ -40,13 +42,14 @@ def read_no_interrupt(stream: IO[Any]) -> bytes | None: except IOError as err: if err.errno != errno.EINTR: raise + return None def process_handler( cmd: Union[str, List[str]], - callback: Callable[[subprocess.Popen], int | str | bytes], + callback: Callable[[subprocess.Popen[bytes]], _T], stderr: int = subprocess.PIPE, -) -> int | str | bytes | None: +) -> _T | None: """Open a command in a shell subprocess and execute a callback. This function provides common scaffolding for creating subprocess.Popen() @@ -137,7 +140,6 @@ def getoutput(cmd: str | list[str]) -> str: out = process_handler(cmd, lambda p: p.communicate()[0], subprocess.STDOUT) if out is None: return '' - assert isinstance(out, bytes) return py3compat.decode(out) @@ -177,10 +179,10 @@ def get_output_error_code(cmd: str | list[str]) -> tuple[str, str, int | None]: returncode: int """ - out_err, p = process_handler(cmd, lambda p: (p.communicate(), p)) - if out_err is None: - return '', '', p.returncode - out, err = out_err + result = process_handler(cmd, lambda p: (p.communicate(), p)) + if result is None: + return '', '', None + (out, err), p = result return py3compat.decode(out), py3compat.decode(err), p.returncode def arg_split(commandline: str, posix: bool = False, strict: bool = True) -> list[str]: @@ -196,7 +198,7 @@ def arg_split(commandline: str, posix: bool = False, strict: bool = True) -> lis command-line args. """ - lex = shlex.shlex(s, posix=posix) + lex = shlex.shlex(commandline, posix=posix) lex.whitespace_split = True # Extract tokens, ensuring that things like leaving open quotes # does not cause this to raise. This is important, because we diff --git a/IPython/utils/_process_win32.py b/IPython/utils/_process_win32.py index ad13cc7a210..1a16fcba1ba 100644 --- a/IPython/utils/_process_win32.py +++ b/IPython/utils/_process_win32.py @@ -65,7 +65,7 @@ def __exit__( os.chdir(self.path) -def _system_body(p: subprocess.Popen) -> int: +def _system_body(p: subprocess.Popen[bytes]) -> int: """Callback for _system.""" enc = DEFAULT_ENCODING @@ -136,9 +136,7 @@ def system(cmd: str) -> Optional[int]: if path is not None: cmd = '"pushd %s &&"%s' % (path, cmd) res = process_handler(cmd, _system_body) - assert isinstance(res, int | type(None)) return res - return None def getoutput(cmd: str) -> str: diff --git a/IPython/utils/py3compat.py b/IPython/utils/py3compat.py index 4f953214737..f503c79257e 100644 --- a/IPython/utils/py3compat.py +++ b/IPython/utils/py3compat.py @@ -10,7 +10,7 @@ from .encoding import DEFAULT_ENCODING -def decode(s, encoding=None): +def decode(s: bytes, encoding: str | None = None) -> str: encoding = encoding or DEFAULT_ENCODING return s.decode(encoding, "replace") diff --git a/pyproject.toml b/pyproject.toml index cfd8a567012..992a0415003 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -130,6 +130,8 @@ enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"] module = [ "IPython.core.events", "IPython.core.magic", + "IPython.utils._process_common", + "IPython.utils._process_win32", ] # Strictest configuration, everything turned up to the max check_untyped_defs = true @@ -302,7 +304,6 @@ module = [ "IPython.terminal.shortcuts.auto_suggest", "IPython.terminal.shortcuts.filters", "IPython.utils._process_cli", - "IPython.utils._process_common", "IPython.utils._process_emscripten", "IPython.utils.capture", "IPython.utils.coloransi",