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

220 statements  

1import sys 

2from collections import OrderedDict 

3from typing import Callable, List, Optional, Union 

4 

5from face.utils import unwrap_text, get_rdep_map, echo 

6from face.errors import ArgumentParseError, CommandLineError, UsageError 

7from face.parser import Parser, Flag, PosArgSpec 

8from face.helpers import HelpHandler 

9from face.middleware import (inject, 

10 get_arg_names, 

11 is_middleware, 

12 face_middleware, 

13 check_middleware, 

14 get_middleware_chain, 

15 _BUILTIN_PROVIDES) 

16 

17from boltons.strutils import camel2under 

18from boltons.iterutils import unique 

19 

20 

21def _get_default_name(func): 

22 from functools import partial 

23 if isinstance(func, partial): 

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

25 

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

27 

28 if ret is None: 

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

30 

31 return ret 

32 

33 

34def _docstring_to_doc(func): 

35 doc = func.__doc__ 

36 if not doc: 

37 return '' 

38 

39 unwrapped = unwrap_text(doc) 

40 try: 

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

42 except IndexError: 

43 ret = '' 

44 

45 return ret 

46 

47 

48def default_print_error(msg): 

49 return echo.err(msg) 

50 

51 

52DEFAULT_HELP_HANDLER = HelpHandler() 

53 

54 

55class CommandGroup: 

56 """A named group of subcommands that are added flat to a parent Command. 

57 

58 When added to a Command via ``cmd.add(group)``, all of the group's 

59 subcommands become direct subcommands of the parent, tagged with the 

60 group name for organized help display. No new subcommand path level 

61 is created. 

62 

63 Args: 

64 name: The group heading displayed in help output. 

65 doc: An optional description (reserved for future use). 

66 """ 

67 def __init__(self, name, doc=None): 

68 if not isinstance(name, str) or not name: 

69 raise ValueError(f'expected non-empty string for CommandGroup name, not: {name!r}') 

70 self.name = name 

71 self.doc = doc 

72 self._commands = [] 

73 

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

75 """Add a subcommand to this group. Accepts the same arguments as 

76 Command() for subcommand construction: a callable (or None) as the 

77 first argument, plus optional name, doc, flags, posargs, etc. 

78 

79 Also accepts a pre-built Command instance directly. 

80 

81 Returns the Command instance. 

82 """ 

83 target = a[0] 

84 if isinstance(target, Command): 

85 subcmd = target 

86 elif callable(target) or target is None: 

87 subcmd = Command(*a, **kw) 

88 else: 

89 raise TypeError(f'expected callable or Command for CommandGroup.add(), not: {target!r}') 

90 self._commands.append(subcmd) 

91 return subcmd 

92 

93 def __repr__(self): 

94 return f'CommandGroup({self.name!r}, commands={len(self._commands)})' 

95 

96 

97# TODO: should name really go here? 

98class Command(Parser): 

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

100 populate it with flags and subcommands, and then call 

101 command.run() to execute your CLI. 

102 

103 Args: 

104 func: The function called when this command is 

105 run with an argv that contains no subcommands. 

106 name: The name of this command, used when this 

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

108 of function) 

109 doc: A description or message that appears in various 

110 help outputs. 

111 flags: A list of Flag instances to initialize the 

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

113 .add() method. 

114 posargs: Pass True if the command takes positional 

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

116 instance. 

117 post_posargs: Pass True if the command takes 

118 additional positional arguments after a conventional '--' 

119 specifier. 

120 help: Pass False to disable the automatically added 

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

122 instance. 

123 middlewares: A list of @face_middleware decorated 

124 callables which participate in dispatch. 

125 group: An optional string group name for display 

126 in help output. See CommandGroup for the recommended way 

127 to group multiple subcommands. 

128 """ 

129 def __init__(self, 

130 func: Optional[Callable], 

131 name: Optional[str] = None, 

132 doc: Optional[str] = None, 

133 *, 

134 flags: Optional[List[Flag]] = None, 

135 posargs: Optional[Union[bool, PosArgSpec]] = None, 

136 post_posargs: Optional[bool] = None, 

137 flagfile: bool = True, 

138 help: Union[bool, HelpHandler] = DEFAULT_HELP_HANDLER, 

139 middlewares: Optional[List[Callable]] = None, 

140 group: Optional[str] = None) -> None: 

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

142 if doc is None: 

143 doc = _docstring_to_doc(func) 

144 

145 # TODO: default posargs if none by inspecting func 

146 super().__init__(name, doc, 

147 flags=flags, 

148 posargs=posargs, 

149 post_posargs=post_posargs, 

150 flagfile=flagfile, 

151 group=group) 

152 

153 self.help_handler = help 

154 

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

156 self._path_func_map = OrderedDict() 

157 self._path_func_map[()] = func 

158 

159 middlewares = list(middlewares or []) 

160 self._path_mw_map = OrderedDict() 

161 self._path_mw_map[()] = [] 

162 self._path_wrapped_map = OrderedDict() 

163 self._path_wrapped_map[()] = func 

164 for mw in middlewares: 

165 self.add_middleware(mw) 

166 

167 if help: 

168 if help is True: 

169 help = DEFAULT_HELP_HANDLER 

170 if help.flag: 

171 self.add(help.flag) 

172 if help.subcmd: 

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

174 

175 if not func and not help: 

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

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

178 

179 return 

180 

181 @property 

182 def func(self): 

183 return self._path_func_map[()] 

184 

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

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

187 

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

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

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

191 and defaults. 

192 

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

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

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

196 more options. 

197 

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

199 exception is only raised on conflicting subcommands and 

200 flags. See add_command for details. 

201 

202 Middleware is only added if it is already decorated with 

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

204 of callables. 

205 

206 """ 

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

208 # conflict 

209 

210 target = a[0] 

211 

212 if isinstance(target, CommandGroup): 

213 return self.add_command_group(target) 

214 

215 if is_middleware(target): 

216 return self.add_middleware(target) 

217 

218 subcmd = a[0] 

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

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

221 

222 if isinstance(subcmd, Command): 

223 self.add_command(subcmd) 

224 return subcmd 

225 

226 flag = a[0] 

227 if not isinstance(flag, Flag): 

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

229 super().add(flag) 

230 

231 return flag 

232 

233 def add_command(self, subcmd): 

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

235 Command. 

236 

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

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

239 conflicting middlewares or subcommand names. 

240 """ 

241 if not isinstance(subcmd, Command): 

242 raise TypeError(f'expected Command instance, not: {subcmd!r}') 

243 self_mw = self._path_mw_map[()] 

244 super().add(subcmd) 

245 # map in new functions 

246 for path in self.subprs_map: 

247 if path not in self._path_func_map: 

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

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

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

251 return 

252 

253 def add_command_group(self, group): 

254 """Add all commands from a CommandGroup as direct subcommands of this 

255 Command, tagged with the group's name for organized help display. 

256 

257 No new subcommand path level is created — the group's commands 

258 become siblings of any existing subcommands. 

259 """ 

260 if not isinstance(group, CommandGroup): 

261 raise TypeError(f'expected CommandGroup instance, not: {group!r}') 

262 for subcmd in group._commands: 

263 subcmd.group = group.name 

264 self.add_command(subcmd) 

265 return 

266 

267 def add_middleware(self, mw): 

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

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

270 

271 """ 

272 if not is_middleware(mw): 

273 mw = face_middleware(mw) 

274 check_middleware(mw) 

275 

276 for flag in mw._face_flags: 

277 self.add(flag) 

278 

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

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

281 

282 return 

283 

284 # TODO: add_flag() 

285 

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

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

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

289 associated subcommand *path*. 

290 """ 

291 flag_map = super().get_flag_map(path=path, with_hidden=with_hidden) 

292 dep_names = self.get_dep_names(path) 

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

294 # the argument parse result and flag dict both capture 

295 # _all_ the flags, so for functions accepting these 

296 # arguments we bypass filtering. 

297 

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

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

300 # this bypassing of filtering will not trigger, unless 

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

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

303 # for middleware. 

304 

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

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

307 return OrderedDict(flag_map) 

308 

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

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

311 

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

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

314 any associated middleware). 

315 

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

317 """ 

318 func = self._path_func_map[path] 

319 if not func: 

320 return [] # for when no handler is specified 

321 

322 mws = self._path_mw_map[path] 

323 

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

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

326 dep_map = {func: set(required_args)} 

327 for mw in mws: 

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

329 for provide in mw._face_provides: 

330 dep_map[provide] = arg_names 

331 if not mw._face_optional: 

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

333 required_args.update(arg_names) 

334 

335 rdep_map = get_rdep_map(dep_map) 

336 

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

338 

339 return sorted(recursive_required_args) 

340 

341 def prepare(self, paths=None): 

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

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

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

345 

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

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

348 conscientious users may want to call this method with no 

349 arguments to validate that all subcommands are ready for 

350 execution. 

351 """ 

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

353 # values are sane there, too 

354 if paths is None: 

355 paths = self._path_func_map.keys() 

356 

357 for path in paths: 

358 func = self._path_func_map[path] 

359 if func is None: 

360 continue # handled by run() 

361 

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

363 provides = [] 

364 if prs.posargs.provides: 

365 provides += [prs.posargs.provides] 

366 if prs.post_posargs.provides: 

367 provides += [prs.post_posargs.provides] 

368 

369 deps = self.get_dep_names(path) 

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

371 all_mws = self._path_mw_map[path] 

372 

373 # filter out unused middlewares 

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

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

376 provides += _BUILTIN_PROVIDES + flag_names 

377 try: 

378 wrapped = get_middleware_chain(mws, func, provides) 

379 except NameError as ne: 

380 ne.args = (ne.args[0] + f' (in path: {path!r})',) 

381 raise 

382 

383 self._path_wrapped_map[path] = wrapped 

384 

385 return 

386 

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

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

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

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

391 caught, a CommandLineError will exit the process, typically 

392 with status code 1. Also handles dispatching to the 

393 appropriate HelpHandler, if configured. 

394 

395 Defaults to handling the arguments on the command line 

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

397 via the *argv* parameter. 

398 

399 Args: 

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

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

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

403 available to the subcommand's handler function. 

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

405 error messages before program exit on CLI errors. 

406 

407 .. note:: 

408 

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

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

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

412 

413 """ 

414 if print_error is None or print_error is True: 

415 print_error = default_print_error 

416 elif print_error and not callable(print_error): 

417 raise TypeError(f'expected callable for print_error, not {print_error!r}') 

418 

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

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

421 

422 try: 

423 prs_res = self.parse(argv=argv) 

424 except ArgumentParseError as ape: 

425 prs_res = ape.prs_res 

426 

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

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

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

430 kwargs.update(prs_res.to_cmd_scope()) 

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

432 

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

434 if prs_res.subcmds: 

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

436 

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

438 # the standard-issue Exception 

439 e_msg = ape.args[0] 

440 if e_msg: 

441 msg += ': ' + e_msg 

442 cle = CommandLineError(msg) 

443 if print_error: 

444 print_error(msg) 

445 raise cle 

446 

447 kwargs.update(prs_res.to_cmd_scope()) 

448 

449 # default in case no middlewares have been installed 

450 func = self._path_func_map[prs_res.subcmds] 

451 

452 cmd = kwargs['subcommand_'] 

453 help_flag_set = (cmd.help_handler 

454 and cmd.help_handler.flag 

455 and prs_res.flags 

456 and prs_res.flags.get(cmd.help_handler.flag.name)) 

457 

458 if help_flag_set: 

459 # Explicit --help: show help, exit 0 

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

461 elif not func: 

462 # No handler (subcommand group invoked without subcommand) 

463 if cmd.help_handler: 

464 inject(cmd.help_handler.func, kwargs) 

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

466 if prs_res.subcmds: 

467 msg += ' ' + ' '.join(prs_res.subcmds) 

468 msg += ': expected a subcommand' 

469 cle = CommandLineError(msg) 

470 if print_error: 

471 print_error(msg) 

472 raise cle 

473 

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

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

476 

477 try: 

478 ret = inject(wrapped, kwargs) 

479 except UsageError as ue: 

480 if print_error: 

481 print_error(ue.format_message()) 

482 raise 

483 return ret