-
Notifications
You must be signed in to change notification settings - Fork 4
/
subprocess.py
63 lines (53 loc) · 2.23 KB
/
subprocess.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
"""Exensions to Python's :py:mod:`subprocess` module.
More specifically, this module provides a customized version of
:py:func:`subprocess.run`, which always sets `check=True`,
`capture_output=True`, enhances the raised exceptions string representation
with additional output information and makes it slightly more readable when
encountered in a stack trace.
"""
from textwrap import indent, wrap
import itertools
import subprocess
class CalledProcessError(subprocess.CalledProcessError):
"""A more verbose version of :py:class:`subprocess.CalledProcessError`.
Replaces the standard string representation of a
:py:class:`subprocess.CalledProcessError` with one that has more output and
error information and is formatted to be more readable in a stack trace.
"""
def __str__(self):
errors = self.stderr.split("\n")
outputs = self.stdout.split("\n")
lines = itertools.chain(
wrap(f"{super().__str__()}"),
["Output:"],
*(
wrap(output, initial_indent=" ", subsequent_indent=" ")
for output in outputs
),
["Errors:"],
*(
wrap(error, initial_indent=" ", subsequent_indent=" ")
for error in errors
),
)
lines = indent("\n".join(lines), "| ")
return f"\n{lines}"
def run(*args, **kwargs):
"""A "safer" version of :py:func:`subprocess.run`.
"Safer" in this context means that this version always raises
:py:class:`CalledProcessError` if the process in question returns a
non-zero exit status. This is done by setting `check=True` and
`capture_output=True`, so you don't have to specify these yourself anymore.
You can though, if you want to override these defaults.
Other than that, the function accepts the same parameters as
:py:func:`subprocess.run`.
"""
for default in ["capture_output", "check", "text"]:
kwargs[default] = kwargs.get(default, True)
try:
result = subprocess.run(*args, **kwargs)
except subprocess.CalledProcessError as cpe:
raise CalledProcessError(
cpe.returncode, cpe.cmd, output=cpe.output, stderr=cpe.stderr
) from None
return result