Coverage for muutils / cli / command.py: 98%

1from __future__ import annotations 

2 

3import os 

4import subprocess 

5import sys 

6from dataclasses import dataclass 

7from typing import Any, List, Union 

8 

9 

10@dataclass 

11class Command: 

12 """Simple typed command with shell flag and subprocess helpers.""" 

13 

14 cmd: Union[List[str], str] 

15 shell: bool = False 

16 env: dict[str, str] | None = None 

17 inherit_env: bool = True 

18 

19 def __post_init__(self) -> None: 

20 """Enforce cmd type when shell is False.""" 

21 if self.shell is False and isinstance(self.cmd, str): 

22 raise ValueError("cmd must be List[str] when shell is False") 

23 

24 def _quote_env(self) -> str: 

25 """Return KEY=VAL tokens for env values. ignores `inherit_env`.""" 

26 if not self.env: 

27 return "" 

28 

29 parts: List[str] = [] 

30 for k, v in self.env.items(): 

31 token: str = f"{k}={v}" 

32 parts.append(token) 

33 prefix: str = " ".join(parts) 

34 return prefix 

35 

36 @property 

37 def cmd_joined(self) -> str: 

38 """Return cmd as a single string, joining with spaces if it's a list. no env included.""" 

39 if isinstance(self.cmd, str): 

40 return self.cmd 

41 else: 

42 return " ".join(self.cmd) 

43 

44 @property 

45 def cmd_for_subprocess(self) -> Union[List[str], str]: 

46 """Return cmd, splitting if shell is True and cmd is a string.""" 

47 if self.shell: 

48 if isinstance(self.cmd, str): 

49 return self.cmd 

50 else: 

51 return " ".join(self.cmd) 

52 else: 

53 assert isinstance(self.cmd, list) 

54 return self.cmd 

55 

56 def script_line(self) -> str: 

57 """Return a single shell string, prefixing KEY=VAL for env if provided.""" 

58 return f"{self._quote_env()} {self.cmd_joined}".strip() 

59 

60 @property 

61 def env_final(self) -> dict[str, str]: 

62 """Return final env dict, merging with os.environ if inherit_env is True.""" 

63 return { 

64 **(os.environ if self.inherit_env else {}), 

65 **(self.env or {}), 

66 } 

67 

68 def run( 

69 self, 

70 **kwargs: Any, 

71 ) -> subprocess.CompletedProcess[Any]: 

72 """Call subprocess.run with this command.""" 

73 try: 

74 return subprocess.run( 

75 self.cmd_for_subprocess, 

76 shell=self.shell, 

77 env=self.env_final, 

78 **kwargs, 

79 ) 

80 except subprocess.CalledProcessError as e: 

81 print(f"Command failed: `{self.script_line()}`", file=sys.stderr) 

82 raise e 

83 

84 def Popen( 

85 self, 

86 **kwargs: Any, 

87 ) -> subprocess.Popen[Any]: 

88 """Call subprocess.Popen with this command.""" 

89 return subprocess.Popen( 

90 self.cmd_for_subprocess, 

91 shell=self.shell, 

92 env=self.env_final, 

93 **kwargs, 

94 )