3 # © Copyright 2021-2022, Scott Gasch
5 """Helper methods concerned with executing subprocesses."""
14 from typing import List, Optional
16 logger = logging.getLogger(__file__)
19 def cmd_showing_output(
22 timeout_seconds: Optional[float] = None,
24 """Kick off a child process. Capture and emit all output that it
25 produces on stdout and stderr in a character by character manner
26 so that we don't have to wait on newlines. This was done to
27 capture the output of a subprocess that created dots to show
28 incremental progress on a task and render it correctly.
31 command: the command to execute
32 timeout_seconds: terminate the subprocess if it takes longer
33 than N seconds; None means to wait as long as it takes.
36 the exit status of the subprocess once the subprocess has
37 exited. Raises TimeoutExpired after killing the subprocess
38 if the timeout expires.
41 prints all output of the child process (stdout or stderr)
46 raise subprocess.TimeoutExpired(command, timeout_seconds)
48 line_enders = set([b'\n', b'\r'])
49 sel = selectors.DefaultSelector()
50 with subprocess.Popen(
54 stdout=subprocess.PIPE,
55 stderr=subprocess.PIPE,
56 universal_newlines=False,
62 timer = threading.Timer(timeout_seconds, timer_expired(p))
65 sel.register(p.stdout, selectors.EVENT_READ) # type: ignore
66 sel.register(p.stderr, selectors.EVENT_READ) # type: ignore
69 for key, _ in sel.select():
70 char = key.fileobj.read(1) # type: ignore
72 sel.unregister(key.fileobj)
73 if len(sel.get_map()) == 0:
78 if key.fileobj is p.stdout:
79 os.write(sys.stdout.fileno(), char)
80 if char in line_enders:
83 os.write(sys.stderr.fileno(), char)
84 if char in line_enders:
93 def cmd_exitcode(command: str, timeout_seconds: Optional[float] = None) -> int:
94 """Run a command silently and return its exit code once it has
95 finished. If timeout_seconds is provided and the command runs too
96 long it will raise a TimeoutExpired exception.
99 command: the command to run
100 timeout_seconds: the max number of seconds to allow the subprocess
101 to execute or None to indicate no timeout
104 the exit status of the subprocess once the subprocess has
107 >>> cmd_exitcode('/bin/echo foo', 10.0)
110 >>> cmd_exitcode('/bin/sleep 2', 0.01)
111 Traceback (most recent call last):
113 subprocess.TimeoutExpired: Command '['/bin/bash', '-c', '/bin/sleep 2']' timed out after 0.01 seconds
116 return subprocess.check_call(["/bin/bash", "-c", command], timeout=timeout_seconds)
119 def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
120 """Run a command and capture its output to stdout and stderr into a
121 string buffer. Return that string as this function's output.
122 Raises subprocess.CalledProcessError or TimeoutExpired on error.
125 command: the command to run
126 timeout_seconds: the max number of seconds to allow the subprocess
127 to execute or None to indicate no timeout
130 The captured output of the subprocess' stdout as a string buffer
132 >>> cmd('/bin/echo foo')[:-1]
135 >>> cmd('/bin/sleep 2', 0.01)
136 Traceback (most recent call last):
138 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.01 seconds
141 ret = subprocess.run(
144 stdout=subprocess.PIPE,
145 stderr=subprocess.STDOUT,
147 timeout=timeout_seconds,
149 return ret.decode("utf-8")
152 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
153 """Run a command silently but raise subprocess.CalledProcessError if
154 it fails and raise a TimeoutExpired if it runs too long.
157 command: the command to run
158 timeout_seconds: the max number of seconds to allow the subprocess
159 to execute or None to indicate no timeout
162 No return value; error conditions (including non-zero child process
163 exits) produce exceptions.
165 >>> run_silently("/usr/bin/true")
167 >>> run_silently("/usr/bin/false")
168 Traceback (most recent call last):
170 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
176 stderr=subprocess.DEVNULL,
177 stdout=subprocess.DEVNULL,
178 capture_output=False,
180 timeout=timeout_seconds,
184 def cmd_in_background(command: str, *, silent: bool = False) -> subprocess.Popen:
185 """Spawns a child process in the background and registers an exit
186 handler to make sure we kill it if the parent process (us) is
190 command: the command to run
191 silent: do not allow any output from the child process to be displayed
192 in the parent process' window
195 the :class:`Popen` object that can be used to communicate
196 with the background process.
198 args = shlex.split(command)
200 subproc = subprocess.Popen(
202 stdin=subprocess.DEVNULL,
203 stdout=subprocess.DEVNULL,
204 stderr=subprocess.DEVNULL,
207 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
209 def kill_subproc() -> None:
211 if subproc.poll() is None:
212 logger.info('At exit handler: killing %s (%s)', subproc, command)
214 subproc.wait(timeout=10.0)
215 except BaseException as be:
218 atexit.register(kill_subproc)
222 def cmd_list(command: List[str]) -> str:
223 """Run a command with args encapsulated in a list and return the
224 output text as a string. Raises subprocess.CalledProcessError.
226 ret = subprocess.run(command, capture_output=True, check=True).stdout
227 return ret.decode("utf-8")
230 if __name__ == '__main__':