1import re
2import subprocess
3import time
4
5_quote_pos = re.compile("(?=[^-0-9a-zA-Z_./\n])")
6
7
8def quote(arg):
9 r"""
10 >>> quote('\t')
11 '\\\t'
12 >>> quote('foo bar')
13 'foo\\ bar'
14 """
15
16 # This is the logic emacs uses
17 if arg:
18 return _quote_pos.sub("\\\\", arg).replace("\n", "'\n'")
19 else:
20 return "''"
21
22
23class ShellCommand:
24 def __init__(self, *args, **kwargs) -> None:
25 self.verbose = kwargs.pop("verbose", False)
26
27 for arg in "stdout", "stderr":
28 if arg not in kwargs:
29 kwargs[arg] = subprocess.PIPE
30
31 self.cmdline = " ".join([quote(x) for x in args])
32
33 if self.verbose:
34 print(f"---- running: {self.cmdline}")
35
36 self.proc = subprocess.run(args, **kwargs)
37
38 def status(self) -> bool:
39 try:
40 self.proc.check_returncode()
41 return True
42 except Exception as e:
43 return False
44
45 def check(self, what: str) -> bool:
46 if self.status():
47 return True
48 else:
49 print(f"==== COMMAND FAILED: {what}")
50 print(f"---- command line: {self.cmdline}")
51 if self.stdout:
52 print("---- stdout ----")
53 print(self.stdout)
54 print("---- end stdout ----")
55 if self.stderr:
56 print("---- stderr ----")
57 print(self.stderr)
58 print("---- end stderr ----")
59
60 return False
61
62 @property
63 def stdout(self) -> str:
64 return self.proc.stdout.decode("utf-8")
65
66 @property
67 def stderr(self) -> str:
68 return self.proc.stderr.decode("utf-8")
69
70 @classmethod
71 def run_with_retry(cls, what: str, *args, **kwargs) -> bool:
72 try_count = 0
73 retries = kwargs.pop("retries", 3)
74 sleep_seconds = kwargs.pop("sleep_seconds", 5)
75 while try_count < retries:
76 if try_count > 0:
77 print(f"Sleeping for {sleep_seconds} before retrying command")
78 time.sleep(sleep_seconds)
79 if cls.run(what, *args, **kwargs):
80 return True
81 try_count += 1
82 return False
83
84 @classmethod
85 def run(cls, what: str, *args, **kwargs) -> bool:
86 return ShellCommand(*args, **kwargs).check(what)
View as plain text