Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/face/command.py: 14%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

183 statements  

1 

2from __future__ import print_function 

3 

4import sys 

5from collections import OrderedDict 

6 

7from face.utils import unwrap_text, get_rdep_map, echo 

8from face.errors import ArgumentParseError, CommandLineError, UsageError 

9from face.parser import Parser, Flag 

10from face.helpers import HelpHandler 

11from face.middleware import (inject, 

12 get_arg_names, 

13 is_middleware, 

14 face_middleware, 

15 check_middleware, 

16 get_middleware_chain, 

17 _BUILTIN_PROVIDES) 

18 

19from boltons.strutils import camel2under 

20from boltons.iterutils import unique 

21 

22 

23def _get_default_name(func): 

24 from functools import partial 

25 if isinstance(func, partial): 

26 func = func.func # just one level of partial for now 

27 

28 # func_name on py2, __name__ on py3 

29 ret = getattr(func, 'func_name', getattr(func, '__name__', None)) # most functions hit this 

30 

31 if ret is None: 

32 ret = camel2under(func.__class__.__name__).lower() # callable instances, etc. 

33 

34 return ret 

35 

36 

37def _docstring_to_doc(func): 

38 doc = func.__doc__ 

39 if not doc: 

40 return '' 

41 

42 unwrapped = unwrap_text(doc) 

43 try: 

44 ret = [g for g in unwrapped.splitlines() if g][0] 

45 except IndexError: 

46 ret = '' 

47 

48 return ret 

49 

50 

51def default_print_error(msg): 

52 return echo.err(msg) 

53 

54 

55DEFAULT_HELP_HANDLER = HelpHandler() 

56 

57 

58# TODO: should name really go here? 

59class Command(Parser): 

60 """The central type in the face framework. Instantiate a Command, 

61 populate it with flags and subcommands, and then call 

62 command.run() to execute your CLI. 

63 

64 Note that only the first three constructor arguments are 

65 positional, the rest are keyword-only. 

66 

67 Args: 

68 func (callable): The function called when this command is 

69 run with an argv that contains no subcommands. 

70 name (str): The name of this command, used when this 

71 command is included as a subcommand. (Defaults to name 

72 of function) 

73 doc (str): A description or message that appears in various 

74 help outputs. 

75 flags (list): A list of Flag instances to initialize the 

76 Command with. Flags can always be added later with the 

77 .add() method. 

78 posargs (bool): Pass True if the command takes positional 

79 arguments. Defaults to False. Can also pass a PosArgSpec 

80 instance. 

81 post_posargs (bool): Pass True if the command takes 

82 additional positional arguments after a conventional '--' 

83 specifier. 

84 help (bool): Pass False to disable the automatically added 

85 --help flag. Defaults to True. Also accepts a HelpHandler 

86 instance, see those docs for more details. 

87 middlewares (list): A list of @face_middleware decorated 

88 callables which participate in dispatch. Also addable 

89 via the .add() method. See Middleware docs for more 

90 details. 

91 

92 """ 

93 def __init__(self, func, name=None, doc=None, **kwargs): 

94 name = name if name is not None else _get_default_name(func) 

95 

96 if doc is None: 

97 doc = _docstring_to_doc(func) 

98 

99 # TODO: default posargs if none by inspecting func 

100 super(Command, self).__init__(name, doc, 

101 flags=kwargs.pop('flags', None), 

102 posargs=kwargs.pop('posargs', None), 

103 post_posargs=kwargs.pop('post_posargs', None), 

104 flagfile=kwargs.pop('flagfile', True)) 

105 

106 _help = kwargs.pop('help', DEFAULT_HELP_HANDLER) 

107 self.help_handler = _help 

108 

109 # TODO: if func is callable, check that "next_" isn't taken 

110 self._path_func_map = OrderedDict() 

111 self._path_func_map[()] = func 

112 

113 middlewares = list(kwargs.pop('middlewares', None) or []) 

114 self._path_mw_map = OrderedDict() 

115 self._path_mw_map[()] = [] 

116 self._path_wrapped_map = OrderedDict() 

117 self._path_wrapped_map[()] = func 

118 for mw in middlewares: 

119 self.add_middleware(mw) 

120 

121 if kwargs: 

122 raise TypeError('unexpected keyword arguments: %r' % sorted(kwargs.keys())) 

123 

124 if _help: 

125 if _help.flag: 

126 self.add(_help.flag) 

127 if _help.subcmd: 

128 self.add(_help.func, _help.subcmd) # for 'help' as a subcmd 

129 

130 if not func and not _help: 

131 raise ValueError('Command requires a handler function or help handler' 

132 ' to be set, not: %r' % func) 

133 

134 return 

135 

136 @property 

137 def func(self): 

138 return self._path_func_map[()] 

139 

140 def add(self, *a, **kw): 

141 """Add a flag, subcommand, or middleware to this Command. 

142 

143 If the first argument is a callable, this method contructs a 

144 Command from it and the remaining arguments, all of which are 

145 optional. See the Command docs for for full details on names 

146 and defaults. 

147 

148 If the first argument is a string, this method constructs a 

149 Flag from that flag string and the rest of the method 

150 arguments, all of which are optional. See the Flag docs for 

151 more options. 

152 

153 If the argument is already an instance of Flag or Command, an 

154 exception is only raised on conflicting subcommands and 

155 flags. See add_command for details. 

156 

157 Middleware is only added if it is already decorated with 

158 @face_middleware. Use .add_middleware() for automatic wrapping 

159 of callables. 

160 

161 """ 

162 # TODO: need to check for middleware provides names + flag names 

163 # conflict 

164 

165 target = a[0] 

166 

167 if is_middleware(target): 

168 return self.add_middleware(target) 

169 

170 subcmd = a[0] 

171 if not isinstance(subcmd, Command) and callable(subcmd) or subcmd is None: 

172 subcmd = Command(*a, **kw) # attempt to construct a new subcmd 

173 

174 if isinstance(subcmd, Command): 

175 self.add_command(subcmd) 

176 return subcmd 

177 

178 flag = a[0] 

179 if not isinstance(flag, Flag): 

180 flag = Flag(*a, **kw) # attempt to construct a Flag from arguments 

181 super(Command, self).add(flag) 

182 

183 return flag 

184 

185 def add_command(self, subcmd): 

186 """Add a Command, and all of its subcommands, as a subcommand of this 

187 Command. 

188 

189 Middleware from the current command is layered on top of the 

190 subcommand's. An exception may be raised if there are 

191 conflicting middlewares or subcommand names. 

192 """ 

193 if not isinstance(subcmd, Command): 

194 raise TypeError('expected Command instance, not: %r' % subcmd) 

195 self_mw = self._path_mw_map[()] 

196 super(Command, self).add(subcmd) 

197 # map in new functions 

198 for path in self.subprs_map: 

199 if path not in self._path_func_map: 

200 self._path_func_map[path] = subcmd._path_func_map[path[1:]] 

201 sub_mw = subcmd._path_mw_map[path[1:]] 

202 self._path_mw_map[path] = self_mw + sub_mw # TODO: check for conflicts 

203 return 

204 

205 def add_middleware(self, mw): 

206 """Add a single middleware to this command. Outermost middleware 

207 should be added first. Remember: first added, first called. 

208 

209 """ 

210 if not is_middleware(mw): 

211 mw = face_middleware(mw) 

212 check_middleware(mw) 

213 

214 for flag in mw._face_flags: 

215 self.add(flag) 

216 

217 for path, mws in self._path_mw_map.items(): 

218 self._path_mw_map[path] = [mw] + mws # TODO: check for conflicts 

219 

220 return 

221 

222 # TODO: add_flag() 

223 

224 def get_flag_map(self, path=(), with_hidden=True): 

225 """Command's get_flag_map differs from Parser's in that it filters 

226 the flag map to just the flags used by the endpoint at the 

227 associated subcommand *path*. 

228 """ 

229 flag_map = super(Command, self).get_flag_map(path=path, with_hidden=with_hidden) 

230 dep_names = self.get_dep_names(path) 

231 if 'args_' in dep_names or 'flags_' in dep_names: 

232 # the argument parse result and flag dict both capture 

233 # _all_ the flags, so for functions accepting these 

234 # arguments we bypass filtering. 

235 

236 # Also note that by setting an argument default in the 

237 # function definition, the dependency becomes "weak", and 

238 # this bypassing of filtering will not trigger, unless 

239 # another function in the chain has a non-default, 

240 # "strong" dependency. This behavior is especially useful 

241 # for middleware. 

242 

243 # TODO: add decorator for the corner case where a function 

244 # accepts these arguments and doesn't use them all. 

245 return OrderedDict(flag_map) 

246 

247 return OrderedDict([(k, f) for k, f in flag_map.items() if f.name in dep_names 

248 or f is self.flagfile_flag or f is self.help_handler.flag]) 

249 

250 def get_dep_names(self, path=()): 

251 """Get a list of the names of all required arguments of a command (and 

252 any associated middleware). 

253 

254 By specifying *path*, the same can be done for any subcommand. 

255 """ 

256 func = self._path_func_map[path] 

257 if not func: 

258 return [] # for when no handler is specified 

259 

260 mws = self._path_mw_map[path] 

261 

262 # start out with all args of handler function, which gets stronger dependencies 

263 required_args = set(get_arg_names(func, only_required=False)) 

264 dep_map = {func: set(required_args)} 

265 for mw in mws: 

266 arg_names = set(get_arg_names(mw, only_required=True)) 

267 for provide in mw._face_provides: 

268 dep_map[provide] = arg_names 

269 if not mw._face_optional: 

270 # all non-optional middlewares get their args required, too. 

271 required_args.update(arg_names) 

272 

273 rdep_map = get_rdep_map(dep_map) 

274 

275 recursive_required_args = rdep_map[func].union(required_args) 

276 

277 return sorted(recursive_required_args) 

278 

279 def prepare(self, paths=None): 

280 """Compile and validate one or more subcommands to ensure all 

281 dependencies are met. Call this once all flags, subcommands, 

282 and middlewares have been added (using .add()). 

283 

284 This method is automatically called by .run() method, but it 

285 only does so for the specific subcommand being invoked. More 

286 conscientious users may want to call this method with no 

287 arguments to validate that all subcommands are ready for 

288 execution. 

289 """ 

290 # TODO: also pre-execute help formatting to make sure all 

291 # values are sane there, too 

292 if paths is None: 

293 paths = self._path_func_map.keys() 

294 

295 for path in paths: 

296 func = self._path_func_map[path] 

297 if func is None: 

298 continue # handled by run() 

299 

300 prs = self.subprs_map[path] if path else self 

301 provides = [] 

302 if prs.posargs.provides: 

303 provides += [prs.posargs.provides] 

304 if prs.post_posargs.provides: 

305 provides += [prs.post_posargs.provides] 

306 

307 deps = self.get_dep_names(path) 

308 flag_names = [f.name for f in self.get_flags(path=path)] 

309 all_mws = self._path_mw_map[path] 

310 

311 # filter out unused middlewares 

312 mws = [mw for mw in all_mws if not mw._face_optional 

313 or [p for p in mw._face_provides if p in deps]] 

314 provides += _BUILTIN_PROVIDES + flag_names 

315 try: 

316 wrapped = get_middleware_chain(mws, func, provides) 

317 except NameError as ne: 

318 ne.args = (ne.args[0] + ' (in path: %r)' % (path,),) 

319 raise 

320 

321 self._path_wrapped_map[path] = wrapped 

322 

323 return 

324 

325 def run(self, argv=None, extras=None, print_error=None): 

326 """Parses arguments and dispatches to the appropriate subcommand 

327 handler. If there is a parse error due to invalid user input, 

328 an error is printed and a CommandLineError is raised. If not 

329 caught, a CommandLineError will exit the process, typically 

330 with status code 1. Also handles dispatching to the 

331 appropriate HelpHandler, if configured. 

332 

333 Defaults to handling the arguments on the command line 

334 (``sys.argv``), but can also be explicitly passed arguments 

335 via the *argv* parameter. 

336 

337 Args: 

338 argv (list): A sequence of strings representing the 

339 command-line arguments. Defaults to ``sys.argv``. 

340 extras (dict): A map of additional arguments to be made 

341 available to the subcommand's handler function. 

342 print_error (callable): The function that formats/prints 

343 error messages before program exit on CLI errors. 

344 

345 .. note:: 

346 

347 For efficiency, :meth:`run()` only checks the subcommand 

348 invoked by *argv*. To ensure that all subcommands are 

349 configured properly, call :meth:`prepare()`. 

350 

351 """ 

352 if print_error is None or print_error is True: 

353 print_error = default_print_error 

354 elif print_error and not callable(print_error): 

355 raise TypeError('expected callable for print_error, not %r' 

356 % print_error) 

357 

358 kwargs = dict(extras) if extras else {} 

359 kwargs['print_error_'] = print_error # TODO: print_error_ in builtin provides? 

360 

361 try: 

362 prs_res = self.parse(argv=argv) 

363 except ArgumentParseError as ape: 

364 prs_res = ape.prs_res 

365 

366 # even if parsing failed, check if the caller was trying to access the help flag 

367 cmd = prs_res.to_cmd_scope()['subcommand_'] 

368 if cmd.help_handler and prs_res.flags and prs_res.flags.get(cmd.help_handler.flag.name): 

369 kwargs.update(prs_res.to_cmd_scope()) 

370 return inject(cmd.help_handler.func, kwargs) 

371 

372 msg = 'error: ' + (prs_res.name or self.name) 

373 if prs_res.subcmds: 

374 msg += ' ' + ' '.join(prs_res.subcmds or ()) 

375 

376 # args attribute, nothing to do with cmdline args this is 

377 # the standard-issue Exception 

378 e_msg = ape.args[0] 

379 if e_msg: 

380 msg += ': ' + e_msg 

381 cle = CommandLineError(msg) 

382 if print_error: 

383 print_error(msg) 

384 raise cle 

385 

386 kwargs.update(prs_res.to_cmd_scope()) 

387 

388 # default in case no middlewares have been installed 

389 func = self._path_func_map[prs_res.subcmds] 

390 

391 cmd = kwargs['subcommand_'] 

392 if cmd.help_handler and (not func or (prs_res.flags and prs_res.flags.get(cmd.help_handler.flag.name))): 

393 return inject(cmd.help_handler.func, kwargs) 

394 elif not func: # pragma: no cover 

395 raise RuntimeError('expected command handler or help handler to be set') 

396 

397 self.prepare(paths=[prs_res.subcmds]) 

398 wrapped = self._path_wrapped_map.get(prs_res.subcmds, func) 

399 

400 try: 

401 ret = inject(wrapped, kwargs) 

402 except UsageError as ue: 

403 if print_error: 

404 print_error(ue.format_message()) 

405 raise 

406 return ret