Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Common utilities for the various process_* implementations.
3This file is only meant to be imported by the platform-specific implementations
4of subprocess utilities, and it contains tools that are common to all of them.
5"""
7#-----------------------------------------------------------------------------
8# Copyright (C) 2010-2011 The IPython Development Team
9#
10# Distributed under the terms of the BSD License. The full license is in
11# the file COPYING, distributed as part of this software.
12#-----------------------------------------------------------------------------
14#-----------------------------------------------------------------------------
15# Imports
16#-----------------------------------------------------------------------------
17from __future__ import annotations
19import os
20import shlex
21import sys
22from typing import IO, TYPE_CHECKING, TypeVar
23from collections.abc import Callable
25if TYPE_CHECKING:
26 import subprocess
28_T = TypeVar("_T")
30from .encoding import DEFAULT_ENCODING
32#-----------------------------------------------------------------------------
33# Function definitions
34#-----------------------------------------------------------------------------
36def read_no_interrupt(stream: IO[bytes]) -> bytes | None:
37 """Read from a pipe ignoring EINTR errors.
39 This is necessary because when reading from pipes with GUI event loops
40 running in the background, often interrupts are raised that stop the
41 command from completing."""
42 import errno
44 try:
45 return stream.read()
46 except OSError as err:
47 if err.errno != errno.EINTR:
48 raise
49 return None
52def process_handler(
53 cmd: str | list[str],
54 callback: Callable[[subprocess.Popen[bytes]], _T],
55 stderr: int | None = None,
56) -> _T | None:
57 """Open a command in a shell subprocess and execute a callback.
59 This function provides common scaffolding for creating subprocess.Popen()
60 calls. It creates a Popen object and then calls the callback with it.
62 Parameters
63 ----------
64 cmd : str or list
65 A command to be executed by the system, using :class:`subprocess.Popen`.
66 If a string is passed, it will be run in the system shell. If a list is
67 passed, it will be used directly as arguments.
68 callback : callable
69 A one-argument function that will be called with the Popen object.
70 stderr : file descriptor number, optional
71 By default this is set to ``subprocess.PIPE``, but you can also pass the
72 value ``subprocess.STDOUT`` to force the subprocess' stderr to go into
73 the same file descriptor as its stdout. This is useful to read stdout
74 and stderr combined in the order they are generated.
76 Returns
77 -------
78 The return value of the provided callback is returned.
79 """
80 import subprocess
82 if stderr is None:
83 stderr = subprocess.PIPE
85 sys.stdout.flush()
86 sys.stderr.flush()
87 # On win32, close_fds can't be true when using pipes for stdin/out/err
88 if sys.platform == "win32" and stderr != subprocess.PIPE:
89 close_fds = False
90 else:
91 close_fds = True
92 # Determine if cmd should be run with system shell.
93 shell = isinstance(cmd, str)
94 # On POSIX systems run shell commands with user-preferred shell.
95 executable = None
96 if shell and os.name == 'posix' and 'SHELL' in os.environ:
97 executable = os.environ['SHELL']
98 p = subprocess.Popen(cmd, shell=shell,
99 executable=executable,
100 stdin=subprocess.PIPE,
101 stdout=subprocess.PIPE,
102 stderr=stderr,
103 close_fds=close_fds)
105 try:
106 out = callback(p)
107 except KeyboardInterrupt:
108 print('^C')
109 sys.stdout.flush()
110 sys.stderr.flush()
111 out = None
112 finally:
113 # Make really sure that we don't leave processes behind, in case the
114 # call above raises an exception
115 # We start by assuming the subprocess finished (to avoid NameErrors
116 # later depending on the path taken)
117 if p.returncode is None:
118 try:
119 p.terminate()
120 p.poll()
121 except OSError:
122 pass
123 # One last try on our way out
124 if p.returncode is None:
125 try:
126 p.kill()
127 except OSError:
128 pass
130 return out
133def getoutput(cmd: str | list[str]) -> str:
134 """Run a command and return its stdout/stderr as a string.
136 Parameters
137 ----------
138 cmd : str or list
139 A command to be executed in the system shell.
141 Returns
142 -------
143 output : str
144 A string containing the combination of stdout and stderr from the
145 subprocess, in whatever order the subprocess originally wrote to its
146 file descriptors (so the order of the information in this string is the
147 correct order as would be seen if running the command in a terminal).
148 """
149 import subprocess
151 out = process_handler(cmd, lambda p: p.communicate()[0], subprocess.STDOUT)
152 if out is None:
153 return ''
154 return out.decode(DEFAULT_ENCODING, "replace")
157def getoutputerror(cmd: str | list[str]) -> tuple[str, str]:
158 """Return (standard output, standard error) of executing cmd in a shell.
160 Accepts the same arguments as os.system().
162 Parameters
163 ----------
164 cmd : str or list
165 A command to be executed in the system shell.
167 Returns
168 -------
169 stdout : str
170 stderr : str
171 """
172 return get_output_error_code(cmd)[:2]
175def get_output_error_code(cmd: str | list[str]) -> tuple[str, str, int | None]:
176 """Return (standard output, standard error, return code) of executing cmd
177 in a shell.
179 Accepts the same arguments as os.system().
181 Parameters
182 ----------
183 cmd : str or list
184 A command to be executed in the system shell.
186 Returns
187 -------
188 stdout : str
189 stderr : str
190 returncode: int
191 """
193 result = process_handler(cmd, lambda p: (p.communicate(), p))
194 if result is None:
195 return '', '', None
196 (out, err), p = result
197 return out.decode(DEFAULT_ENCODING, "replace"), err.decode(DEFAULT_ENCODING, "replace"), p.returncode
199def arg_split(commandline: str, posix: bool = False, strict: bool = True) -> list[str]:
200 """Split a command line's arguments in a shell-like manner.
202 This is a modified version of the standard library's shlex.split()
203 function, but with a default of posix=False for splitting, so that quotes
204 in inputs are respected.
206 if strict=False, then any errors shlex.split would raise will result in the
207 unparsed remainder being the last element of the list, rather than raising.
208 This is because we sometimes use arg_split to parse things other than
209 command-line args.
210 """
212 lex = shlex.shlex(commandline, posix=posix)
213 lex.whitespace_split = True
214 # Extract tokens, ensuring that things like leaving open quotes
215 # does not cause this to raise. This is important, because we
216 # sometimes pass Python source through this (e.g. %timeit f(" ")),
217 # and it shouldn't raise an exception.
218 # It may be a bad idea to parse things that are not command-line args
219 # through this function, but we do, so let's be safe about it.
220 lex.commenters='' #fix for GH-1269
221 tokens = []
222 while True:
223 try:
224 tokens.append(next(lex))
225 except StopIteration:
226 break
227 except ValueError:
228 if strict:
229 raise
230 # couldn't parse, get remaining blob as last token
231 tokens.append(lex.token)
232 break
234 return tokens
237def arg_split_with_quotes(
238 commandline: str, strict: bool = True
239) -> list[tuple[str, bool]]:
240 """Split a command line and report which tokens were originally quoted.
242 Returns a list of ``(token, was_quoted)`` pairs. ``token`` is the unquoted
243 form, as ``shlex.split(posix=True)`` returns, and ``was_quoted`` is True
244 if that token had any single- or double-quote characters in ``commandline``.
246 Useful for callers like ``%run`` that want to honor shell quoting when
247 deciding whether to apply further expansion (glob, tilde) to a token.
249 Detection is shlex-based on both passes so the quote semantics are the
250 same on Posix and Windows. If ``strict`` is False, malformed input (e.g.
251 an unbalanced quote) returns whatever was parsed so far instead of raising.
252 """
253 def _tokenize(s: str, posix: bool) -> list[str]:
254 lex = shlex.shlex(s, posix=posix)
255 lex.whitespace_split = True
256 lex.commenters = ''
257 out = []
258 while True:
259 try:
260 out.append(next(lex))
261 except StopIteration:
262 break
263 except ValueError:
264 if strict:
265 raise
266 out.append(lex.token)
267 break
268 return out
270 raw_tokens = _tokenize(commandline, posix=False)
271 clean_tokens = _tokenize(commandline, posix=True)
273 if len(raw_tokens) != len(clean_tokens):
274 # If the two passes disagree (exotic input) report nothing as quoted
275 # so callers get the legacy, non-quote-aware behavior.
276 return [(t, False) for t in clean_tokens]
278 return [
279 (clean, ("'" in raw) or ('"' in raw))
280 for clean, raw in zip(clean_tokens, raw_tokens)
281 ]