Add type annotations to _process_common.py and _process_win32.py (#15134)

Carreau · web-flow · commit 1123b01fb21c · 2026-02-19T13:20:30.000+01:00
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
diff --git 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
@@ -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:
+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:
     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],
-    stderr=subprocess.PIPE,
-) -> int | str | bytes:
+    callback: Callable[[subprocess.Popen[bytes]], _T],
+    stderr: int = subprocess.PIPE,
+) -> _T | None:
     """Open a command in a shell subprocess and execute a callback.
 
     This function provides common scaffolding for creating subprocess.Popen()
@@ -118,7 +121,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
@@ -137,11 +140,10 @@ def getoutput(cmd):
     out = process_handler(cmd, lambda p: p.communicate()[0], subprocess.STDOUT)
     if out is None:
         return ''
-    assert isinstance(out, bytes)
     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 +160,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.
 
@@ -176,13 +179,13 @@ def get_output_error_code(cmd):
     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(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()
@@ -195,7 +198,7 @@ def arg_split(s, posix=False, strict=True):
     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
@@ -59,13 +59,13 @@ 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)
 
 
-def _system_body(p: subprocess.Popen) -> int:
+def _system_body(p: subprocess.Popen[bytes]) -> int:
     """Callback for _system."""
     enc = DEFAULT_ENCODING
 
@@ -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:
@@ -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:
@@ -210,6 +208,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
diff --git 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
@@ -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",
