Coverage for pycommons / processes / shell.py: 98%
92 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-24 03:11 +0000
1r"""
2The tool for invoking shell commands.
4>>> Command(("echo", "123"), stdout=STREAM_CAPTURE).execute(False)
5('123\n', None)
6"""
8import subprocess # nosec
9from dataclasses import dataclass
10from os import getcwd
11from typing import Callable, Final, Iterable, Mapping
13from pycommons.io.console import logger
14from pycommons.io.path import UTF8, Path, directory_path
15from pycommons.types import check_int_range, type_error
17#: ignore the given stream
18STREAM_IGNORE: Final[int] = 0
19#: forward given stream to the same stream of this process
20STREAM_FORWARD: Final[int] = 1
21#: capture the given stream
22STREAM_CAPTURE: Final[int] = 2
25#: the stream mode to string converter
26_SM: Final[Callable[[int], str]] = {
27 STREAM_IGNORE: " ignored",
28 STREAM_FORWARD: " forwarded",
29 STREAM_CAPTURE: " captured",
30}.get
33@dataclass(frozen=True, init=False, order=False, eq=False)
34class Command:
35 """
36 A class that represents a command that can be executed.
38 >>> c = Command("test")
39 >>> c.command
40 ('test',)
41 >>> c.working_dir.is_dir()
42 True
43 >>> c.timeout
44 3600
46 >>> d = Command(("test", "b"))
47 >>> d.command
48 ('test', 'b')
49 >>> d.working_dir == c.working_dir
50 True
51 >>> d.timeout == c.timeout
52 True
54 >>> e = Command(("", "test", " b", " "))
55 >>> e.command == d.command
56 True
57 >>> e.working_dir == c.working_dir
58 True
59 >>> e.timeout == c.timeout
60 True
62 >>> try:
63 ... Command(1)
64 ... except TypeError as te:
65 ... print(str(te)[:50])
66 command should be an instance of any in {str, typi
68 >>> try:
69 ... Command([1])
70 ... except TypeError as te:
71 ... print(te)
72 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
74 >>> try:
75 ... Command(["x", 1])
76 ... except TypeError as te:
77 ... print(te)
78 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
80 >>> try:
81 ... Command([])
82 ... except ValueError as ve:
83 ... print(ve)
84 Invalid command [].
86 >>> try:
87 ... Command([""])
88 ... except ValueError as ve:
89 ... print(ve)
90 Invalid command [''].
92 >>> try:
93 ... Command("")
94 ... except ValueError as ve:
95 ... print(ve)
96 Invalid command [''].
98 >>> Command("x", working_dir=Path(__file__).up(1)).command
99 ('x',)
101 >>> try:
102 ... Command("x", working_dir=1)
103 ... except TypeError as te:
104 ... print(te)
105 descriptor '__len__' requires a 'str' object but received a 'int'
107 >>> try:
108 ... Command("x", working_dir=Path(__file__))
109 ... except ValueError as ve:
110 ... print(str(ve)[-30:])
111 does not identify a directory.
113 >>> Command("x", timeout=23).timeout
114 23
116 >>> try:
117 ... Command("x", timeout=1.2)
118 ... except TypeError as te:
119 ... print(te)
120 timeout should be an instance of int but is float, namely 1.2.
122 >>> try:
123 ... Command("x", timeout=None)
124 ... except TypeError as te:
125 ... print(te)
126 timeout should be an instance of int but is None.
128 >>> try:
129 ... Command("x", timeout=0)
130 ... except ValueError as ve:
131 ... print(ve)
132 timeout=0 is invalid, must be in 1..1000000.
134 >>> try:
135 ... Command("x", timeout=1_000_001)
136 ... except ValueError as ve:
137 ... print(ve)
138 timeout=1000001 is invalid, must be in 1..1000000.
140 >>> try:
141 ... Command("x", stdin=1_000_001)
142 ... except TypeError as te:
143 ... print(str(te)[:49])
144 stdin should be an instance of any in {None, str}
146 >>> sxx = str(Command("x", env={"A": "B", "C": "D"}))
147 >>> sxx[sxx.index("with "):sxx.index("with ") + 30]
148 'with <env> no stdin, stdout ig'
150 >>> try:
151 ... Command("x", env={"A": "B", "C": 1})
152 ... except TypeError as te:
153 ... print(str(te))
154 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
156 >>> try:
157 ... Command("x", env=1)
158 ... except TypeError as te:
159 ... print(str(te)[:-20])
160 env should be an instance of any in {typing.Iterable, typing.Mapping} b
162 >>> str(Command("x", env=dict()))[0:10]
163 "('x',) in "
164 """
166 #: the command line.
167 command: tuple[str, ...]
168 #: the working directory
169 working_dir: Path
170 #: the timeout in seconds, after which the process will be terminated
171 timeout: int
172 #: the data to be written to stdin
173 stdin: str | None
174 #: how to handle the standard output stream
175 stdout: int
176 #: how to handle the standard error stream
177 stderr: int
178 #: the environment variables to pass to the new process, if any
179 env: tuple[tuple[str, str], ...] | None
181 def __init__(self, command: str | Iterable[str],
182 working_dir: str | None = None,
183 timeout: int | None = 3600,
184 stdin: str | None = None,
185 stdout: int = STREAM_IGNORE,
186 stderr: int = STREAM_IGNORE,
187 env: Mapping[str, str] | Iterable[tuple[
188 str, str]] | None = None) -> None:
189 """
190 Create the command.
192 :param command: the command string or iterable
193 :param working_dir: the working directory
194 :param timeout: the timeout
195 :param stdin: a string to be written to stdin, or `None`
196 :param stdout: how to handle the standard output stream
197 :param stderr: how to handle the standard error stream
198 """
199 if isinstance(command, str):
200 command = [command]
201 elif not isinstance(command, Iterable):
202 raise type_error(command, "command", (str, Iterable))
203 object.__setattr__(self, "command", tuple(
204 s for s in map(str.strip, command) if str.__len__(s) > 0))
205 if tuple.__len__(self.command) <= 0:
206 raise ValueError(f"Invalid command {command!r}.")
208 object.__setattr__(self, "working_dir", directory_path(
209 getcwd() if working_dir is None else working_dir))
211 object.__setattr__(self, "timeout", check_int_range(
212 timeout, "timeout", 1, 1_000_000))
214 if (stdin is not None) and (not isinstance(stdin, str)):
215 raise type_error(stdin, "stdin", (str, None))
216 object.__setattr__(self, "stdin", stdin)
218 object.__setattr__(self, "stdout", check_int_range(
219 stdout, "stdout", 0, 2))
220 object.__setattr__(self, "stderr", check_int_range(
221 stderr, "stderr", 0, 2))
223 the_env: tuple[tuple[str, str], ...] | None = None
224 if env is not None:
225 if isinstance(env, Mapping):
226 env = env.items()
227 elif not isinstance(env, Iterable):
228 raise type_error(env, "env", (Mapping, Iterable))
229 the_env = tuple(sorted((str.strip(k), str.strip(v))
230 for k, v in env))
231 if tuple.__len__(the_env) <= 0:
232 the_env = None
233 object.__setattr__(self, "env", the_env)
235 def __str__(self) -> str:
236 """
237 Get the string representation of this command.
239 This string includes most of the important information, but it will
240 never include the environment variables. These variables may contain
241 security sensitive stuff, so they are not printed. Instead. if an
242 environment is specified, this will just be printed as `<env>`.
244 :returns: A string representing this command
246 >>> str(Command("a"))[-50:]
247 ' with no stdin, stdout ignored, and stderr ignored'
248 >>> str(Command("x"))[:11]
249 "('x',) in '"
250 >>> "with 3 chars of stdin" in str(Command("x", stdin="123"))
251 True
252 >>> str(Command("a", env={"y": "x"}))[-65:]
253 'for 3600s with <env> no stdin, stdout ignored, and stderr ignored'
254 """
255 si: str = "no" if self.stdin is None \
256 else f"{str.__len__(self.stdin)} chars of"
257 ev: str = "" if self.env is None else "<env> "
258 return (f"{self.command!r} in {self.working_dir!r} for {self.timeout}"
259 f"s with {ev}{si} stdin, stdout{_SM(self.stdout)}, and "
260 f"stderr{_SM(self.stderr)}")
262 def execute(self, log_call: bool = True) -> tuple[str | None, str | None]:
263 r"""
264 Execute the given process.
266 :param log_call: should the call be logged? If `True`, the
267 string representation of the :class:`Command` will be
268 written to the `logger`, otherwise nothing is logged.
269 Note: The environment, if any, will not be printed for security
270 reasons.
271 :returns: a tuple with the standard output and standard error, which
272 are only not `None` if they were supposed to be captured
273 :raises TypeError: if any argument has the wrong type
274 :raises ValueError: if execution of the process failed
276 >>> Command(("echo", "123"), stdout=STREAM_CAPTURE).execute(False)
277 ('123\n', None)
279 >>> Command(("echo", "", "123"), stdout=STREAM_CAPTURE).execute(False)
280 ('123\n', None)
282 >>> from contextlib import redirect_stdout
283 >>> with redirect_stdout(None):
284 ... s = Command(("echo", "123"), stdout=STREAM_CAPTURE).execute()
285 >>> print(s)
286 ('123\n', None)
288 >>> Command("cat", stdin="test", stdout=STREAM_CAPTURE).execute(False)
289 ('test', None)
291 >>> Command("cat", stdin="test").execute(False)
292 (None, None)
294 >>> try:
295 ... with redirect_stdout(None):
296 ... Command(("ping", "blabla!")).execute(True)
297 ... except ValueError as ve:
298 ... ss = str(ve)
299 ... print(ss[:20] + " ... " + ss[-22:])
300 ('ping', 'blabla!') ... yields return code 2.
302 >>> try:
303 ... with redirect_stdout(None):
304 ... Command(("ping", "www.example.com", "-i 20"),
305 ... timeout=1).execute(True)
306 ... except ValueError as ve:
307 ... print("timed out after" in str(ve))
308 True
310 >>> try:
311 ... Command("x").execute(None)
312 ... except TypeError as te:
313 ... print(te)
314 log_call should be an instance of bool but is None.
316 >>> try:
317 ... Command("x").execute(1)
318 ... except TypeError as te:
319 ... print(te)
320 log_call should be an instance of bool but is int, namely 1.
322 >>> with redirect_stdout(None):
323 ... r = Command(("echo", "1"), stderr=STREAM_CAPTURE).execute(
324 ... True)
325 >>> r
326 (None, '')
328 >>> with redirect_stdout(None):
329 ... r = Command(("printenv", ),
330 ... stdout=STREAM_CAPTURE,
331 ... env={"BLA": "XX"}).execute(True)
332 >>> r
333 ('BLA=XX\n', None)
334 """
335 if not isinstance(log_call, bool):
336 raise type_error(log_call, "log_call", bool)
337 message: Final[str] = str(self)
338 if log_call:
339 logger(f"Now invoking {message}.")
341 arguments: Final[dict[str, str | Iterable[str] | bool | int]] = {
342 "args": self.command,
343 "check": False,
344 "text": True,
345 "timeout": self.timeout,
346 "cwd": self.working_dir,
347 "errors": "strict",
348 "encoding": UTF8,
349 }
351 if self.stdin is not None:
352 arguments["input"] = self.stdin
354 arguments["stdout"] = 1 if self.stdout == STREAM_FORWARD else (
355 subprocess.PIPE if self.stdout == STREAM_CAPTURE
356 else subprocess.DEVNULL)
357 arguments["stderr"] = 2 if self.stderr == STREAM_FORWARD else (
358 subprocess.PIPE if self.stderr == STREAM_CAPTURE
359 else subprocess.DEVNULL)
361 if self.env is not None:
362 arguments["env"] = {t[0]: t[1] for t in self.env}
364 try:
365 # noqa # nosemgrep # pylint: disable=W1510 # type: ignore
366 ret: Final[subprocess.CompletedProcess] = \
367 subprocess.run(**arguments) # type: ignore # nosec # noqa
368 except (TimeoutError, subprocess.TimeoutExpired) as toe:
369 if log_call:
370 logger(f"Failed executing {self} with timeout {toe}.")
371 raise ValueError(f"{message} timed out: {toe}.") from toe
373 returncode: Final[int] = ret.returncode
374 if returncode != 0:
375 if log_call:
376 logger(f"Failed executing {self}: got return"
377 f" code {returncode}.")
378 raise ValueError(f"{message} yields return code {returncode}.")
380 stdout: str | None = None
381 if self.stdout == STREAM_CAPTURE:
382 stdout = ret.stdout
383 if not isinstance(stdout, str):
384 raise type_error(stdout, f"stdout of {self}", stdout)
386 stderr: str | None = None
387 if self.stderr == STREAM_CAPTURE:
388 stderr = ret.stderr
389 if not isinstance(stderr, str):
390 raise type_error(stderr, f"stderr of {self}", stderr)
392 if log_call:
393 capture: str = ""
394 if stdout is not None:
395 capture = f", captured {str.__len__(stdout)} chars of stdout"
396 if stderr is not None:
397 capture = f"{capture} and " if str.__len__(capture) > 0 \
398 else ", captured "
399 capture = f"{capture}{str.__len__(stderr)} chars of stderr"
400 logger(f"Finished executing {self} with return code 0{capture}.")
402 return stdout, stderr