Coverage for .tox/cov/lib/python3.10/site-packages/confattr/types.py: 79%

1#!./runmodule.sh 

2 

3import os 

4import shutil 

5import shlex 

6import typing 

7from collections.abc import Sequence, Callable, Mapping, MutableMapping 

8 

9from .subprocess_pipe import run_and_pipe, CompletedProcess 

10 

11 

12TYPE_CONTEXT: 'typing.TypeAlias' = 'Callable[[SubprocessCommand], typing.ContextManager[SubprocessCommand]] | None' 

13 

14 

15class Path: 

16 

17 ''' 

18 This is the path as it is stored in the config file. 

19 It needs to be processed before usage. 

20 In the easiest case that is as easy as calling os.path.expanduser 

21 but you may want to do more like checking that the path exists 

22 or mounting an external drive. 

23 ''' 

24 

25 type_name = 'path' 

26 help = 'The path to a file or directory' 

27 

28 def __init__(self, value: str) -> None: 

29 self.raw = value 

30 

31 def __str__(self) -> str: 

32 return self.raw 

33 

34 def __repr__(self) -> str: 

35 return '%s(%r)' % (type(self).__name__, self.raw) 

36 

37 def __bool__(self) -> bool: 

38 return bool(self.raw) 

39 

40 

41class Regex(str): 

42 

43 type_name = 'regular expression' 

44 help = ''' 

45 A regular expression in python syntax. 

46 You can specify flags by starting the regular expression with `(?aiLmsux)`. 

47 https://docs.python.org/3/library/re.html#regular-expression-syntax 

48 ''' 

49 

50 

51class SubprocessCommand: 

52 

53 type_name = 'command' 

54 help = '''\ 

55 A command to be executed as a subprocess. 

56 The command is executed without a shell so redirection or wildcard expansion is not possible. 

57 Setting environment variables and piping like in a POSIX shell, however, are implemented in python and should work platform independently. 

58 If you need a shell write the command to a file, insert an appropriate shebang line, make the file executable and set this value to the file. 

59 ''' 

60 

61 python_callbacks: 'MutableMapping[str, Callable[[SubprocessCommand, TYPE_CONTEXT], None]]' = {} 

62 

63 @classmethod 

64 def register_python_callback(cls, name: str, func: 'Callable[[SubprocessCommand, TYPE_CONTEXT], None]') -> None: 

65 cls.python_callbacks[name] = func 

66 

67 @classmethod 

68 def unregister_python_callback(cls, name: str) -> None: 

69 del cls.python_callbacks[name] 

70 

71 @classmethod 

72 def has_python_callback(cls, name: str) -> bool: 

73 return name in cls.python_callbacks 

74 

75 

76 def __init__(self, arg: 'SubprocessCommand|Sequence[str]|str', *, env: 'Mapping[str, str]|None' = None) -> None: 

77 self.cmd: 'Sequence[str]' 

78 self.env: 'Mapping[str, str]|None' 

79 if isinstance(arg, str): 

80 assert env is None 

81 self.parse_str(arg) 

82 elif isinstance(arg, SubprocessCommand): 

83 self.cmd = list(arg.cmd) 

84 self.env = dict(arg.env) if arg.env else None 

85 if env: 

86 if self.env: 

87 self.env.update(env) 

88 else: 

89 self.env = env 

90 else: 

91 self.cmd = list(arg) 

92 self.env = env 

93 

94 def parse_str(self, arg: str) -> None: 

95 ''' 

96 Parses a string as returned by :meth:`__str__` and initializes this objcet accordingly 

97 

98 :param arg: The string to be parsed 

99 :raises ValueError: if arg is invalid 

100 

101 Example: 

102 If the input is ``arg = 'ENVVAR1=val ENVVAR2= cmd --arg1 --arg2'`` 

103 this function sets 

104 .. code-block:: 

105 

106 self.env = {'ENVVAR1' : 'val', 'ENVVAR2' : ''} 

107 self.cmd = ['cmd', '--arg1', '--arg2'] 

108 ''' 

109 if not arg: 

110 raise ValueError('cmd is empty') 

111 

112 cmd = shlex.split(arg) 

113 

114 self.env = {} 

115 for i in range(len(cmd)): 

116 if '=' in cmd[i]: 

117 var, val = cmd[i].split('=', 1) 

118 self.env[var] = val 

119 else: 

120 self.cmd = cmd[i:] 

121 if not self.env: 

122 self.env = None 

123 return 

124 

125 raise ValueError('cmd consists of environment variables only, there is no command to be executed') 

126 

127 # ------- compare ------- 

128 

129 def __eq__(self, other: typing.Any) -> bool: 

130 if isinstance(other, SubprocessCommand): 

131 return self.cmd == other.cmd and self.env == other.env 

132 return NotImplemented 

133 

134 # ------- custom methods ------- 

135 

136 def replace(self, wildcard: str, replacement: str) -> 'SubprocessCommand': 

137 return SubprocessCommand([replacement if word == wildcard else word for word in self.cmd], env=self.env) 

138 

139 def run(self, *, context: 'TYPE_CONTEXT|None') -> 'CompletedProcess[bytes]|None': 

140 ''' 

141 Runs this command and returns when the command is finished. 

142 

143 :param context: returns a context manager which can be used to stop and start an urwid screen. 

144 It takes the command to be executed so that it can log the command 

145 and it returns the command to be executed so that it can modify the command, 

146 e.g. processing and intercepting some environment variables. 

147 

148 :return: The completed process 

149 :raises OSError: e.g. if the program was not found 

150 :raises CalledProcessError: if the called program failed 

151 ''' 

152 if self.cmd[0] in self.python_callbacks: 

153 self.python_callbacks[self.cmd[0]](self, context) 

154 return None 

155 

156 if context is None: 

157 return run_and_pipe(self.cmd, env=self._add_os_environ(self.env)) 

158 

159 with context(self) as command: 

160 return run_and_pipe(command.cmd, env=self._add_os_environ(command.env)) 

161 

162 @staticmethod 

163 def _add_os_environ(env: 'Mapping[str, str]|None') -> 'Mapping[str, str]|None': 

164 if env is None: 

165 return env 

166 return dict(os.environ, **env) 

167 

168 def is_installed(self) -> bool: 

169 return self.cmd[0] in self.python_callbacks or bool(shutil.which(self.cmd[0])) 

170 

171 # ------- to str ------- 

172 

173 def __str__(self) -> str: 

174 if self.env: 

175 env = ' '.join('%s=%s' % (var, shlex.quote(val)) for var, val in self.env.items()) 

176 env += ' ' 

177 else: 

178 env = '' 

179 return env + ' '.join(shlex.quote(word) for word in self.cmd) 

180 

181 def __repr__(self) -> str: 

182 return '%s(%r, env=%r)' % (type(self).__name__, self.cmd, self.env) 

183 

184class SubprocessCommandWithAlternatives: 

185 

186 type_name = 'command with alternatives' 

187 help = ''' 

188 One or more commands separated by ||. 

189 The first command where the program is installed is executed. The other commands are ignored. 

190 

191 The command is executed without a shell so redirection or wildcard expansion is not possible. 

192 Setting environment variables and piping like in a POSIX shell, however, are implemented in python and should work platform independently. 

193 If you need a shell write the command to a file, insert an appropriate shebang line, make the file executable and set this value to the file. 

194 ''' 

195 

196 SEP = '||' 

197 

198 def get_preferred_command(self) -> SubprocessCommand: 

199 for cmd in self.commands: 

200 if cmd.is_installed(): 

201 return cmd 

202 

203 raise FileNotFoundError('none of the commands is installed: %s' % self) 

204 

205 

206 def __init__(self, commands: 'Sequence[SubprocessCommand|Sequence[str]|str]|str') -> None: 

207 if isinstance(commands, str): 

208 self.commands = [SubprocessCommand(cmd) for cmd in commands.split(self.SEP)] 

209 else: 

210 self.commands = [SubprocessCommand(cmd) for cmd in commands] 

211 

212 

213 def __str__(self) -> str: 

214 return self.SEP.join(str(cmd) for cmd in self.commands) 

215 

216 def __repr__(self) -> str: 

217 return '%s(%s)' % (type(self).__name__, self.commands) 

218 

219 

220 def replace(self, wildcard: str, replacement: str) -> SubprocessCommand: 

221 return self.get_preferred_command().replace(wildcard, replacement) 

222 

223 def run(self, context: 'TYPE_CONTEXT|None' = None) -> 'CompletedProcess[bytes]|None': 

224 return self.get_preferred_command().run(context=context)