3 # © Copyright 2021-2023, 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 raw, character by character,
26 manner so that we don't have to wait on newlines. This was done
27 to capture, for example, the output of a subprocess that creates
28 dots to show incremental progress on a task and render it
32 command: the command to execute
33 timeout_seconds: terminate the subprocess if it takes longer
34 than N seconds; None means to wait as long as it takes.
37 the exit status of the subprocess once the subprocess has
38 exited. Raises `TimeoutExpired` after killing the subprocess
39 if the timeout expires.
42 TimeoutExpired: if timeout expires before child terminates
45 prints all output of the child process (stdout or stderr)
50 raise subprocess.TimeoutExpired(command, timeout_seconds)
52 line_enders = set([b"\n", b"\r"])
53 sel = selectors.DefaultSelector()
54 with subprocess.Popen(
58 stdout=subprocess.PIPE,
59 stderr=subprocess.PIPE,
60 universal_newlines=False,
66 timer = threading.Timer(timeout_seconds, timer_expired(p))
69 sel.register(p.stdout, selectors.EVENT_READ) # type: ignore
70 sel.register(p.stderr, selectors.EVENT_READ) # type: ignore
73 for key, _ in sel.select():
74 char = key.fileobj.read(1) # type: ignore
76 sel.unregister(key.fileobj)
82 if key.fileobj is p.stdout:
83 os.write(sys.stdout.fileno(), char)
84 if char in line_enders:
87 os.write(sys.stderr.fileno(), char)
88 if char in line_enders:
97 def cmd_exitcode(command: str, timeout_seconds: Optional[float] = None) -> int:
98 """Run a command silently in the background and return its exit
99 code once it has finished.
102 command: the command to run
103 timeout_seconds: optional the max number of seconds to allow
104 the subprocess to execute or None to indicate no timeout
107 the exit status of the subprocess once the subprocess has
111 TimeoutExpired: if timeout_seconds is provided and the child process
112 executes longer than the limit.
114 >>> cmd_exitcode('/bin/echo foo', 10.0)
117 >>> cmd_exitcode('/bin/sleep 2', 0.01)
118 Traceback (most recent call last):
120 subprocess.TimeoutExpired: Command '['/bin/bash', '-c', '/bin/sleep 2']' timed out after 0.01 seconds
123 return subprocess.check_call(["/bin/bash", "-c", command], timeout=timeout_seconds)
126 def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
127 """Run a command and capture its output to stdout and stderr into a
128 string buffer. Return that string as this function's output.
131 command: the command to run
132 timeout_seconds: the max number of seconds to allow the subprocess
133 to execute or None to indicate no timeout
136 The captured output of the subprocess' stdout as a string buffer
139 CalledProcessError: the child process didn't exit cleanly
140 TimeoutExpired: the child process ran too long
142 >>> cmd('/bin/echo foo')[:-1]
145 >>> cmd('/bin/sleep 2', 0.01)
146 Traceback (most recent call last):
148 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.01 seconds
151 ret = subprocess.run(
154 stdout=subprocess.PIPE,
155 stderr=subprocess.STDOUT,
157 timeout=timeout_seconds,
159 return ret.decode("utf-8")
162 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
163 """Run a command silently.
166 command: the command to run.
167 timeout_seconds: the optional max number of seconds to allow
168 the subprocess to execute or None (default) to indicate no
172 No return value; error conditions (including non-zero child process
173 exits) produce exceptions.
176 CalledProcessError: if the child process fails (i.e. exit != 0)
177 TimeoutExpired: if the child process executes too long.
179 >>> run_silently("/usr/bin/true")
181 >>> run_silently("/usr/bin/false")
182 Traceback (most recent call last):
184 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
190 stderr=subprocess.DEVNULL,
191 stdout=subprocess.DEVNULL,
192 capture_output=False,
194 timeout=timeout_seconds,
198 def cmd_in_background(command: str, *, silent: bool = False) -> subprocess.Popen:
199 """Spawns a child process in the background and registers an exit
200 handler to make sure we kill it if the parent process (us) is
204 command: the command to run
205 silent: do not allow any output from the child process to be displayed
206 in the parent process' window
209 the :class:`Popen` object that can be used to communicate
210 with the background process.
212 args = shlex.split(command)
214 subproc = subprocess.Popen(
216 stdin=subprocess.DEVNULL,
217 stdout=subprocess.DEVNULL,
218 stderr=subprocess.DEVNULL,
221 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
223 def kill_subproc() -> None:
225 if subproc.poll() is None:
226 logger.info("At exit handler: killing %s (%s)", subproc, command)
228 subproc.wait(timeout=10.0)
229 except BaseException:
231 "Failed to terminate background process %s; giving up.", subproc
234 atexit.register(kill_subproc)
238 def cmd_list(command: List[str]) -> str:
239 """Run a command with args encapsulated in a list and return the
240 output text as a string.
243 CalledProcessError: the child process didn't exit cleanly
244 TimeoutExpired: the child process ran too long
246 ret = subprocess.run(command, capture_output=True, check=True).stdout
247 return ret.decode("utf-8")
250 if __name__ == "__main__":