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.
129 Raises subprocess.CalledProcessError or TimeoutExpired on error.
132 command: the command to run
133 timeout_seconds: the max number of seconds to allow the subprocess
134 to execute or None to indicate no timeout
137 The captured output of the subprocess' stdout as a string buffer
139 >>> cmd('/bin/echo foo')[:-1]
142 >>> cmd('/bin/sleep 2', 0.01)
143 Traceback (most recent call last):
145 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.01 seconds
148 ret = subprocess.run(
151 stdout=subprocess.PIPE,
152 stderr=subprocess.STDOUT,
154 timeout=timeout_seconds,
156 return ret.decode("utf-8")
159 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
160 """Run a command silently.
163 command: the command to run.
164 timeout_seconds: the optional max number of seconds to allow
165 the subprocess to execute or None (default) to indicate no
169 No return value; error conditions (including non-zero child process
170 exits) produce exceptions.
173 CalledProcessError: if the child process fails (i.e. exit != 0)
174 TimeoutExpired: if the child process executes too long.
176 >>> run_silently("/usr/bin/true")
178 >>> run_silently("/usr/bin/false")
179 Traceback (most recent call last):
181 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
187 stderr=subprocess.DEVNULL,
188 stdout=subprocess.DEVNULL,
189 capture_output=False,
191 timeout=timeout_seconds,
195 def cmd_in_background(command: str, *, silent: bool = False) -> subprocess.Popen:
196 """Spawns a child process in the background and registers an exit
197 handler to make sure we kill it if the parent process (us) is
201 command: the command to run
202 silent: do not allow any output from the child process to be displayed
203 in the parent process' window
206 the :class:`Popen` object that can be used to communicate
207 with the background process.
209 args = shlex.split(command)
211 subproc = subprocess.Popen(
213 stdin=subprocess.DEVNULL,
214 stdout=subprocess.DEVNULL,
215 stderr=subprocess.DEVNULL,
218 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
220 def kill_subproc() -> None:
222 if subproc.poll() is None:
223 logger.info("At exit handler: killing %s (%s)", subproc, command)
225 subproc.wait(timeout=10.0)
226 except BaseException:
228 "Failed to terminate background process %s; giving up.", subproc
231 atexit.register(kill_subproc)
235 def cmd_list(command: List[str]) -> str:
236 """Run a command with args encapsulated in a list and return the
237 output text as a string. Raises subprocess.CalledProcessError.
239 ret = subprocess.run(command, capture_output=True, check=True).stdout
240 return ret.decode("utf-8")
243 if __name__ == "__main__":