Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
one more pass
  • Loading branch information
Carreau committed Feb 19, 2026
commit a1076b0178ac20655bb3c4bdadca4e963255a25b
22 changes: 12 additions & 10 deletions IPython/utils/_process_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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()
Expand Down Expand Up @@ -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)


Expand Down Expand Up @@ -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]:
Expand All @@ -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
Expand Down
4 changes: 1 addition & 3 deletions IPython/utils/_process_win32.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion IPython/utils/py3compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Expand Down
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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",
Expand Down
Loading