Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/program.py: 17%

1import getpass 

2import inspect 

3import json 

4import os 

5import sys 

6import textwrap 

7from importlib import import_module # buffalo buffalo 

8from typing import ( 

9 TYPE_CHECKING, 

10 Any, 

11 Dict, 

12 List, 

13 Optional, 

14 Sequence, 

15 Tuple, 

16 Type, 

17) 

18 

19from . import Collection, Config, Executor, FilesystemLoader 

20from .completion.complete import complete, print_completion_script 

21from .parser import Parser, ParserContext, Argument 

22from .exceptions import UnexpectedExit, CollectionNotFound, ParseError, Exit 

23from .terminals import pty_size 

24from .util import debug, enable_logging, helpline 

25 

26if TYPE_CHECKING: 

27 from .loader import Loader 

28 from .parser import ParseResult 

29 from .util import Lexicon 

30 

31 

32class Program: 

33 """ 

34 Manages top-level CLI invocation, typically via ``setup.py`` entrypoints. 

35 

36 Designed for distributing Invoke task collections as standalone programs, 

37 but also used internally to implement the ``invoke`` program itself. 

38 

39 .. seealso:: 

40 :ref:`reusing-as-a-binary` for a tutorial/walkthrough of this 

41 functionality. 

42 

43 .. versionadded:: 1.0 

44 """ 

45 

46 core: "ParseResult" 

47 

48 def core_args(self) -> List["Argument"]: 

49 """ 

50 Return default core `.Argument` objects, as a list. 

51 

52 .. versionadded:: 1.0 

53 """ 

54 # Arguments present always, even when wrapped as a different binary 

55 return [ 

56 Argument( 

57 names=("command-timeout", "T"), 

58 kind=int, 

59 help="Specify a global command execution timeout, in seconds.", 

60 ), 

61 Argument( 

62 names=("complete",), 

63 kind=bool, 

64 default=False, 

65 help="Print tab-completion candidates for given parse remainder.", # noqa 

66 ), 

67 Argument( 

68 names=("config", "f"), 

69 help="Runtime configuration file to use.", 

70 ), 

71 Argument( 

72 names=("debug", "d"), 

73 kind=bool, 

74 default=False, 

75 help="Enable debug output.", 

76 ), 

77 Argument( 

78 names=("dry", "R"), 

79 kind=bool, 

80 default=False, 

81 help="Echo commands instead of running.", 

82 ), 

83 Argument( 

84 names=("echo", "e"), 

85 kind=bool, 

86 default=False, 

87 help="Echo executed commands before running.", 

88 ), 

89 Argument( 

90 names=("help", "h"), 

91 optional=True, 

92 help="Show core or per-task help and exit.", 

93 ), 

94 Argument( 

95 names=("hide",), 

96 help="Set default value of run()'s 'hide' kwarg.", 

97 ), 

98 Argument( 

99 names=("list", "l"), 

100 optional=True, 

101 help="List available tasks, optionally limited to a namespace.", # noqa 

102 ), 

103 Argument( 

104 names=("list-depth", "D"), 

105 kind=int, 

106 default=0, 

107 help="When listing tasks, only show the first INT levels.", 

108 ), 

109 Argument( 

110 names=("list-format", "F"), 

111 help="Change the display format used when listing tasks. Should be one of: flat (default), nested, json.", # noqa 

112 default="flat", 

113 ), 

114 Argument( 

115 names=("print-completion-script",), 

116 kind=str, 

117 default="", 

118 help="Print the tab-completion script for your preferred shell (bash|zsh|fish).", # noqa 

119 ), 

120 Argument( 

121 names=("prompt-for-sudo-password",), 

122 kind=bool, 

123 default=False, 

124 help="Prompt user at start of session for the sudo.password config value.", # noqa 

125 ), 

126 Argument( 

127 names=("pty", "p"), 

128 kind=bool, 

129 default=False, 

130 help="Use a pty when executing shell commands.", 

131 ), 

132 Argument( 

133 names=("version", "V"), 

134 kind=bool, 

135 default=False, 

136 help="Show version and exit.", 

137 ), 

138 Argument( 

139 names=("warn-only", "w"), 

140 kind=bool, 

141 default=False, 

142 help="Warn, instead of failing, when shell commands fail.", 

143 ), 

144 Argument( 

145 names=("write-pyc",), 

146 kind=bool, 

147 default=False, 

148 help="Enable creation of .pyc files.", 

149 ), 

150 ] 

151 

152 def task_args(self) -> List["Argument"]: 

153 """ 

154 Return default task-related `.Argument` objects, as a list. 

155 

156 These are only added to the core args in "task runner" mode (the 

157 default for ``invoke`` itself) - they are omitted when the constructor 

158 is given a non-empty ``namespace`` argument ("bundled namespace" mode). 

159 

160 .. versionadded:: 1.0 

161 """ 

162 # Arguments pertaining specifically to invocation as 'invoke' itself 

163 # (or as other arbitrary-task-executing programs, like 'fab') 

164 return [ 

165 Argument( 

166 names=("collection", "c"), 

167 help="Specify collection name to load.", 

168 ), 

169 Argument( 

170 names=("no-dedupe",), 

171 kind=bool, 

172 default=False, 

173 help="Disable task deduplication.", 

174 ), 

175 Argument( 

176 names=("search-root", "r"), 

177 help="Change root directory used for finding task modules.", 

178 ), 

179 ] 

180 

181 argv: List[str] 

182 # Other class-level global variables a subclass might override sometime 

183 # maybe? 

184 leading_indent_width = 2 

185 leading_indent = " " * leading_indent_width 

186 indent_width = 4 

187 indent = " " * indent_width 

188 col_padding = 3 

189 

190 def __init__( 

191 self, 

192 version: Optional[str] = None, 

193 namespace: Optional["Collection"] = None, 

194 name: Optional[str] = None, 

195 binary: Optional[str] = None, 

196 loader_class: Optional[Type["Loader"]] = None, 

197 executor_class: Optional[Type["Executor"]] = None, 

198 config_class: Optional[Type["Config"]] = None, 

199 binary_names: Optional[List[str]] = None, 

200 ) -> None: 

201 """ 

202 Create a new, parameterized `.Program` instance. 

203 

204 :param str version: 

205 The program's version, e.g. ``"0.1.0"``. Defaults to ``"unknown"``. 

206 

207 :param namespace: 

208 A `.Collection` to use as this program's subcommands. 

209 

210 If ``None`` (the default), the program will behave like ``invoke``, 

211 seeking a nearby task namespace with a `.Loader` and exposing 

212 arguments such as :option:`--list` and :option:`--collection` for 

213 inspecting or selecting specific namespaces. 

214 

215 If given a `.Collection` object, will use it as if it had been 

216 handed to :option:`--collection`. Will also update the parser to 

217 remove references to tasks and task-related options, and display 

218 the subcommands in ``--help`` output. The result will be a program 

219 that has a static set of subcommands. 

220 

221 :param str name: 

222 The program's name, as displayed in ``--version`` output. 

223 

224 If ``None`` (default), is a capitalized version of the first word 

225 in the ``argv`` handed to `.run`. For example, when invoked from a 

226 binstub installed as ``foobar``, it will default to ``Foobar``. 

227 

228 :param str binary: 

229 Descriptive lowercase binary name string used in help text. 

230 

231 For example, Invoke's own internal value for this is ``inv[oke]``, 

232 denoting that it is installed as both ``inv`` and ``invoke``. As 

233 this is purely text intended for help display, it may be in any 

234 format you wish, though it should match whatever you've put into 

235 your ``setup.py``'s ``console_scripts`` entry. 

236 

237 If ``None`` (default), uses the first word in ``argv`` verbatim (as 

238 with ``name`` above, except not capitalized). 

239 

240 :param binary_names: 

241 List of binary name strings, for use in completion scripts. 

242 

243 This list ensures that the shell completion scripts generated by 

244 :option:`--print-completion-script` instruct the shell to use 

245 that completion for all of this program's installed names. 

246 

247 For example, Invoke's internal default for this is ``["inv", 

248 "invoke"]``. 

249 

250 If ``None`` (the default), the first word in ``argv`` (in the 

251 invocation of :option:`--print-completion-script`) is used in a 

252 single-item list. 

253 

254 :param loader_class: 

255 The `.Loader` subclass to use when loading task collections. 

256 

257 Defaults to `.FilesystemLoader`. 

258 

259 :param executor_class: 

260 The `.Executor` subclass to use when executing tasks. 

261 

262 Defaults to `.Executor`; may also be overridden at runtime by the 

263 :ref:`configuration system <default-values>` and its 

264 ``tasks.executor_class`` setting (anytime that setting is not 

265 ``None``). 

266 

267 :param config_class: 

268 The `.Config` subclass to use for the base config object. 

269 

270 Defaults to `.Config`. 

271 

272 .. versionchanged:: 1.2 

273 Added the ``binary_names`` argument. 

274 """ 

275 self.version = "unknown" if version is None else version 

276 self.namespace = namespace 

277 self._name = name 

278 # TODO 3.0: rename binary to binary_help_name or similar. (Or write 

279 # code to autogenerate it from binary_names.) 

280 self._binary = binary 

281 self._binary_names = binary_names 

282 self.argv = [] 

283 self.loader_class = loader_class or FilesystemLoader 

284 self.executor_class = executor_class or Executor 

285 self.config_class = config_class or Config 

286 

287 def create_config(self) -> None: 

288 """ 

289 Instantiate a `.Config` (or subclass, depending) for use in task exec. 

290 

291 This Config is fully usable but will lack runtime-derived data like 

292 project & runtime config files, CLI arg overrides, etc. That data is 

293 added later in `update_config`. See `.Config` docstring for lifecycle 

294 details. 

295 

296 :returns: ``None``; sets ``self.config`` instead. 

297 

298 .. versionadded:: 1.0 

299 """ 

300 self.config = self.config_class() 

301 

302 def update_config(self, merge: bool = True) -> None: 

303 """ 

304 Update the previously instantiated `.Config` with parsed data. 

305 

306 For example, this is how ``--echo`` is able to override the default 

307 config value for ``run.echo``. 

308 

309 :param bool merge: 

310 Whether to merge at the end, or defer. Primarily useful for 

311 subclassers. Default: ``True``. 

312 

313 .. versionadded:: 1.0 

314 """ 

315 # Now that we have parse results handy, we can grab the remaining 

316 # config bits: 

317 # - runtime config, as it is dependent on the runtime flag/env var 

318 # - the overrides config level, as it is composed of runtime flag data 

319 # NOTE: only fill in values that would alter behavior, otherwise we 

320 # want the defaults to come through. 

321 run = {} 

322 if self.args["warn-only"].value: 

323 run["warn"] = True 

324 if self.args.pty.value: 

325 run["pty"] = True 

326 if self.args.hide.value: 

327 run["hide"] = self.args.hide.value 

328 if self.args.echo.value: 

329 run["echo"] = True 

330 if self.args.dry.value: 

331 run["dry"] = True 

332 tasks = {} 

333 if "no-dedupe" in self.args and self.args["no-dedupe"].value: 

334 tasks["dedupe"] = False 

335 timeouts = {} 

336 command = self.args["command-timeout"].value 

337 if command: 

338 timeouts["command"] = command 

339 # Handle "fill in config values at start of runtime", which for now is 

340 # just sudo password 

341 sudo = {} 

342 if self.args["prompt-for-sudo-password"].value: 

343 prompt = "Desired 'sudo.password' config value: " 

344 sudo["password"] = getpass.getpass(prompt) 

345 overrides = dict(run=run, tasks=tasks, sudo=sudo, timeouts=timeouts) 

346 self.config.load_overrides(overrides, merge=False) 

347 runtime_path = self.args.config.value 

348 if runtime_path is None: 

349 runtime_path = os.environ.get("INVOKE_RUNTIME_CONFIG", None) 

350 self.config.set_runtime_path(runtime_path) 

351 self.config.load_runtime(merge=False) 

352 if merge: 

353 self.config.merge() 

354 

355 def run(self, argv: Optional[List[str]] = None, exit: bool = True) -> None: 

356 """ 

357 Execute main CLI logic, based on ``argv``. 

358 

359 :param argv: 

360 The arguments to execute against. May be ``None``, a list of 

361 strings, or a string. See `.normalize_argv` for details. 

362 

363 :param bool exit: 

364 When ``False`` (default: ``True``), will ignore `.ParseError`, 

365 `.Exit` and `.Failure` exceptions, which otherwise trigger calls to 

366 `sys.exit`. 

367 

368 .. note:: 

369 This is mostly a concession to testing. If you're setting this 

370 to ``False`` in a production setting, you should probably be 

371 using `.Executor` and friends directly instead! 

372 

373 .. versionadded:: 1.0 

374 """ 

375 try: 

376 # Create an initial config, which will hold defaults & values from 

377 # most config file locations (all but runtime.) Used to inform 

378 # loading & parsing behavior. 

379 self.create_config() 

380 # Parse the given ARGV with our CLI parsing machinery, resulting in 

381 # things like self.args (core args/flags), self.collection (the 

382 # loaded namespace, which may be affected by the core flags) and 

383 # self.tasks (the tasks requested for exec and their own 

384 # args/flags) 

385 self.parse_core(argv) 

386 # Handle collection concerns including project config 

387 self.parse_collection() 

388 # Parse remainder of argv as task-related input 

389 self.parse_tasks() 

390 # End of parsing (typically bailout stuff like --list, --help) 

391 self.parse_cleanup() 

392 # Update the earlier Config with new values from the parse step - 

393 # runtime config file contents and flag-derived overrides (e.g. for 

394 # run()'s echo, warn, etc options.) 

395 self.update_config() 

396 # Create an Executor, passing in the data resulting from the prior 

397 # steps, then tell it to execute the tasks. 

398 self.execute() 

399 except (UnexpectedExit, Exit, ParseError) as e: 

400 debug("Received a possibly-skippable exception: {!r}".format(e)) 

401 # Print error messages from parser, runner, etc if necessary; 

402 # prevents messy traceback but still clues interactive user into 

403 # problems. 

404 if isinstance(e, ParseError): 

405 print(e, file=sys.stderr) 

406 if isinstance(e, Exit) and e.message: 

407 print(e.message, file=sys.stderr) 

408 if isinstance(e, UnexpectedExit) and e.result.hide: 

409 print(e, file=sys.stderr, end="") 

410 # Terminate execution unless we were told not to. 

411 if exit: 

412 if isinstance(e, UnexpectedExit): 

413 code = e.result.exited 

414 elif isinstance(e, Exit): 

415 code = e.code 

416 elif isinstance(e, ParseError): 

417 code = 1 

418 sys.exit(code) 

419 else: 

420 debug("Invoked as run(..., exit=False), ignoring exception") 

421 except KeyboardInterrupt: 

422 sys.exit(1) # Same behavior as Python itself outside of REPL 

423 

424 def parse_core(self, argv: Optional[List[str]]) -> None: 

425 debug("argv given to Program.run: {!r}".format(argv)) 

426 self.normalize_argv(argv) 

427 

428 # Obtain core args (sets self.core) 

429 self.parse_core_args() 

430 debug("Finished parsing core args") 

431 

432 # Set interpreter bytecode-writing flag 

433 sys.dont_write_bytecode = not self.args["write-pyc"].value 

434 

435 # Enable debugging from here on out, if debug flag was given. 

436 # (Prior to this point, debugging requires setting INVOKE_DEBUG). 

437 if self.args.debug.value: 

438 enable_logging() 

439 

440 # Short-circuit if --version 

441 if self.args.version.value: 

442 debug("Saw --version, printing version & exiting") 

443 self.print_version() 

444 raise Exit 

445 

446 # Print (dynamic, no tasks required) completion script if requested 

447 if self.args["print-completion-script"].value: 

448 print_completion_script( 

449 shell=self.args["print-completion-script"].value, 

450 names=self.binary_names, 

451 ) 

452 raise Exit 

453 

454 def parse_collection(self) -> None: 

455 """ 

456 Load a tasks collection & project-level config. 

457 

458 .. versionadded:: 1.0 

459 """ 

460 # Load a collection of tasks unless one was already set. 

461 if self.namespace is not None: 

462 debug( 

463 "Program was given default namespace, not loading collection" 

464 ) 

465 self.collection = self.namespace 

466 else: 

467 debug( 

468 "No default namespace provided, trying to load one from disk" 

469 ) # noqa 

470 # If no bundled namespace & --help was given, just print it and 

471 # exit. (If we did have a bundled namespace, core --help will be 

472 # handled *after* the collection is loaded & parsing is done.) 

473 if self.args.help.value is True: 

474 debug( 

475 "No bundled namespace & bare --help given; printing help." 

476 ) 

477 self.print_help() 

478 raise Exit 

479 self.load_collection() 

480 # Set these up for potential use later when listing tasks 

481 # TODO: be nice if these came from the config...! Users would love to 

482 # say they default to nested for example. Easy 2.x feature-add. 

483 self.list_root: Optional[str] = None 

484 self.list_depth: Optional[int] = None 

485 self.list_format = "flat" 

486 self.scoped_collection = self.collection 

487 

488 # TODO: load project conf, if possible, gracefully 

489 

490 def parse_cleanup(self) -> None: 

491 """ 

492 Post-parsing, pre-execution steps such as --help, --list, etc. 

493 

494 .. versionadded:: 1.0 

495 """ 

496 halp = self.args.help.value 

497 

498 # Core (no value given) --help output (only when bundled namespace) 

499 if halp is True: 

500 debug("Saw bare --help, printing help & exiting") 

501 self.print_help() 

502 raise Exit 

503 

504 # Print per-task help, if necessary 

505 if halp: 

506 if halp in self.parser.contexts: 

507 msg = "Saw --help <taskname>, printing per-task help & exiting" 

508 debug(msg) 

509 self.print_task_help(halp) 

510 raise Exit 

511 else: 

512 # TODO: feels real dumb to factor this out of Parser, but...we 

513 # should? 

514 raise ParseError("No idea what '{}' is!".format(halp)) 

515 

516 # Print discovered tasks if necessary 

517 list_root = self.args.list.value # will be True or string 

518 self.list_format = self.args["list-format"].value 

519 self.list_depth = self.args["list-depth"].value 

520 if list_root: 

521 # Not just --list, but --list some-root - do moar work 

522 if isinstance(list_root, str): 

523 self.list_root = list_root 

524 try: 

525 sub = self.collection.subcollection_from_path(list_root) 

526 self.scoped_collection = sub 

527 except KeyError: 

528 msg = "Sub-collection '{}' not found!" 

529 raise Exit(msg.format(list_root)) 

530 self.list_tasks() 

531 raise Exit 

532 

533 # Print completion helpers if necessary 

534 if self.args.complete.value: 

535 complete( 

536 names=self.binary_names, 

537 core=self.core, 

538 initial_context=self.initial_context, 

539 collection=self.collection, 

540 # NOTE: can't reuse self.parser as it has likely been mutated 

541 # between when it was set and now. 

542 parser=self._make_parser(), 

543 ) 

544 

545 # Fallback behavior if no tasks were given & no default specified 

546 # (mostly a subroutine for overriding purposes) 

547 # NOTE: when there is a default task, Executor will select it when no 

548 # tasks were found in CLI parsing. 

549 if not self.tasks and not self.collection.default: 

550 self.no_tasks_given() 

551 

552 def no_tasks_given(self) -> None: 

553 debug( 

554 "No tasks specified for execution and no default task; printing global help as fallback" # noqa 

555 ) 

556 self.print_help() 

557 raise Exit 

558 

559 def execute(self) -> None: 

560 """ 

561 Hand off data & tasks-to-execute specification to an `.Executor`. 

562 

563 .. note:: 

564 Client code just wanting a different `.Executor` subclass can just 

565 set ``executor_class`` in `.__init__`, or override 

566 ``tasks.executor_class`` anywhere in the :ref:`config system 

567 <default-values>` (which may allow you to avoid using a custom 

568 Program entirely). 

569 

570 .. versionadded:: 1.0 

571 """ 

572 klass = self.executor_class 

573 config_path = self.config.tasks.executor_class 

574 if config_path is not None: 

575 # TODO: why the heck is this not builtin to importlib? 

576 module_path, _, class_name = config_path.rpartition(".") 

577 # TODO: worth trying to wrap both of these and raising ImportError 

578 # for cases where module exists but class name does not? More 

579 # "normal" but also its own possible source of bugs/confusion... 

580 module = import_module(module_path) 

581 klass = getattr(module, class_name) 

582 executor = klass(self.collection, self.config, self.core) 

583 executor.execute(*self.tasks) 

584 

585 def normalize_argv(self, argv: Optional[List[str]]) -> None: 

586 """ 

587 Massages ``argv`` into a useful list of strings. 

588 

589 **If None** (the default), uses `sys.argv`. 

590 

591 **If a non-string iterable**, uses that in place of `sys.argv`. 

592 

593 **If a string**, performs a `str.split` and then executes with the 

594 result. (This is mostly a convenience; when in doubt, use a list.) 

595 

596 Sets ``self.argv`` to the result. 

597 

598 .. versionadded:: 1.0 

599 """ 

600 if argv is None: 

601 argv = sys.argv 

602 debug("argv was None; using sys.argv: {!r}".format(argv)) 

603 elif isinstance(argv, str): 

604 argv = argv.split() 

605 debug("argv was string-like; splitting: {!r}".format(argv)) 

606 self.argv = argv 

607 

608 @property 

609 def name(self) -> str: 

610 """ 

611 Derive program's human-readable name based on `.binary`. 

612 

613 .. versionadded:: 1.0 

614 """ 

615 return self._name or self.binary.capitalize() 

616 

617 @property 

618 def called_as(self) -> str: 

619 """ 

620 Returns the program name we were actually called as. 

621 

622 Specifically, this is the (Python's os module's concept of a) basename 

623 of the first argument in the parsed argument vector. 

624 

625 .. versionadded:: 1.2 

626 """ 

627 # XXX: defaults to empty string if 'argv' is '[]' or 'None' 

628 return os.path.basename(self.argv[0]) if self.argv else "" 

629 

630 @property 

631 def binary(self) -> str: 

632 """ 

633 Derive program's help-oriented binary name(s) from init args & argv. 

634 

635 .. versionadded:: 1.0 

636 """ 

637 return self._binary or self.called_as 

638 

639 @property 

640 def binary_names(self) -> List[str]: 

641 """ 

642 Derive program's completion-oriented binary name(s) from args & argv. 

643 

644 .. versionadded:: 1.2 

645 """ 

646 return self._binary_names or [self.called_as] 

647 

648 # TODO 3.0: ugh rename this or core_args, they are too confusing 

649 @property 

650 def args(self) -> "Lexicon": 

651 """ 

652 Obtain core program args from ``self.core`` parse result. 

653 

654 .. versionadded:: 1.0 

655 """ 

656 return self.core[0].args 

657 

658 @property 

659 def initial_context(self) -> ParserContext: 

660 """ 

661 The initial parser context, aka core program flags. 

662 

663 The specific arguments contained therein will differ depending on 

664 whether a bundled namespace was specified in `.__init__`. 

665 

666 .. versionadded:: 1.0 

667 """ 

668 args = self.core_args() 

669 if self.namespace is None: 

670 args += self.task_args() 

671 return ParserContext(args=args) 

672 

673 def print_version(self) -> None: 

674 print("{} {}".format(self.name, self.version or "unknown")) 

675 

676 def print_help(self) -> None: 

677 usage_suffix = "task1 [--task1-opts] ... taskN [--taskN-opts]" 

678 if self.namespace is not None: 

679 usage_suffix = "<subcommand> [--subcommand-opts] ..." 

680 print("Usage: {} [--core-opts] {}".format(self.binary, usage_suffix)) 

681 print("") 

682 print("Core options:") 

683 print("") 

684 self.print_columns(self.initial_context.help_tuples()) 

685 if self.namespace is not None: 

686 self.list_tasks() 

687 

688 def parse_core_args(self) -> None: 

689 """ 

690 Filter out core args, leaving any tasks or their args for later. 

691 

692 Sets ``self.core`` to the `.ParseResult` from this step. 

693 

694 .. versionadded:: 1.0 

695 """ 

696 debug("Parsing initial context (core args)") 

697 parser = Parser(initial=self.initial_context, ignore_unknown=True) 

698 self.core = parser.parse_argv(self.argv[1:]) 

699 msg = "Core-args parse result: {!r} & unparsed: {!r}" 

700 debug(msg.format(self.core, self.core.unparsed)) 

701 

702 def load_collection(self) -> None: 

703 """ 

704 Load a task collection based on parsed core args, or die trying. 

705 

706 .. versionadded:: 1.0 

707 """ 

708 # NOTE: start, coll_name both fall back to configuration values within 

709 # Loader (which may, however, get them from our config.) 

710 start = self.args["search-root"].value 

711 loader = self.loader_class( # type: ignore 

712 config=self.config, start=start 

713 ) 

714 coll_name = self.args.collection.value 

715 try: 

716 module, parent = loader.load(coll_name) 

717 # This is the earliest we can load project config, so we should - 

718 # allows project config to affect the task parsing step! 

719 # TODO: is it worth merging these set- and load- methods? May 

720 # require more tweaking of how things behave in/after __init__. 

721 self.config.set_project_location(parent) 

722 self.config.load_project() 

723 self.collection = Collection.from_module( 

724 module, 

725 loaded_from=parent, 

726 auto_dash_names=self.config.tasks.auto_dash_names, 

727 ) 

728 except CollectionNotFound as e: 

729 raise Exit("Can't find any collection named {!r}!".format(e.name)) 

730 

731 def _update_core_context( 

732 self, context: ParserContext, new_args: Dict[str, Any] 

733 ) -> None: 

734 # Update core context w/ core_via_task args, if and only if the 

735 # via-task version of the arg was truly given a value. 

736 # TODO: push this into an Argument-aware Lexicon subclass and 

737 # .update()? 

738 for key, arg in new_args.items(): 

739 if arg.got_value: 

740 context.args[key]._value = arg._value 

741 

742 def _make_parser(self) -> Parser: 

743 return Parser( 

744 initial=self.initial_context, 

745 contexts=self.collection.to_contexts( 

746 ignore_unknown_help=self.config.tasks.ignore_unknown_help 

747 ), 

748 ) 

749 

750 def parse_tasks(self) -> None: 

751 """ 

752 Parse leftover args, which are typically tasks & per-task args. 

753 

754 Sets ``self.parser`` to the parser used, ``self.tasks`` to the 

755 parsed per-task contexts, and ``self.core_via_tasks`` to a context 

756 holding any core flags seen within the task contexts. 

757 

758 Also modifies ``self.core`` to include the data from ``core_via_tasks`` 

759 (so that it correctly reflects any supplied core flags regardless of 

760 where they appeared). 

761 

762 .. versionadded:: 1.0 

763 """ 

764 self.parser = self._make_parser() 

765 debug("Parsing tasks against {!r}".format(self.collection)) 

766 result = self.parser.parse_argv(self.core.unparsed) 

767 self.core_via_tasks = result.pop(0) 

768 self._update_core_context( 

769 context=self.core[0], new_args=self.core_via_tasks.args 

770 ) 

771 self.tasks = result 

772 debug("Resulting task contexts: {!r}".format(self.tasks)) 

773 

774 def print_task_help(self, name: str) -> None: 

775 """ 

776 Print help for a specific task, e.g. ``inv --help <taskname>``. 

777 

778 .. versionadded:: 1.0 

779 """ 

780 # Setup 

781 ctx = self.parser.contexts[name] 

782 tuples = ctx.help_tuples() 

783 docstring = inspect.getdoc(self.collection[name]) 

784 header = "Usage: {} [--core-opts] {} {}[other tasks here ...]" 

785 opts = "[--options] " if tuples else "" 

786 print(header.format(self.binary, name, opts)) 

787 print("") 

788 print("Docstring:") 

789 if docstring: 

790 # Really wish textwrap worked better for this. 

791 for line in docstring.splitlines(): 

792 if line.strip(): 

793 print(self.leading_indent + line) 

794 else: 

795 print("") 

796 print("") 

797 else: 

798 print(self.leading_indent + "none") 

799 print("") 

800 print("Options:") 

801 if tuples: 

802 self.print_columns(tuples) 

803 else: 

804 print(self.leading_indent + "none") 

805 print("") 

806 

807 def list_tasks(self) -> None: 

808 # Short circuit if no tasks to show (Collection now implements bool) 

809 focus = self.scoped_collection 

810 if not focus: 

811 msg = "No tasks found in collection '{}'!" 

812 raise Exit(msg.format(focus.name)) 

813 # TODO: now that flat/nested are almost 100% unified, maybe rethink 

814 # this a bit? 

815 getattr(self, "list_{}".format(self.list_format))() 

816 

817 def list_flat(self) -> None: 

818 pairs = self._make_pairs(self.scoped_collection) 

819 self.display_with_columns(pairs=pairs) 

820 

821 def list_nested(self) -> None: 

822 pairs = self._make_pairs(self.scoped_collection) 

823 extra = "'*' denotes collection defaults" 

824 self.display_with_columns(pairs=pairs, extra=extra) 

825 

826 def _make_pairs( 

827 self, 

828 coll: "Collection", 

829 ancestors: Optional[List[str]] = None, 

830 ) -> List[Tuple[str, Optional[str]]]: 

831 if ancestors is None: 

832 ancestors = [] 

833 pairs = [] 

834 indent = len(ancestors) * self.indent 

835 ancestor_path = ".".join(x for x in ancestors) 

836 for name, task in sorted(coll.tasks.items()): 

837 is_default = name == coll.default 

838 # Start with just the name and just the aliases, no prefixes or 

839 # dots. 

840 displayname = name 

841 aliases = list(map(coll.transform, sorted(task.aliases))) 

842 # If displaying a sub-collection (or if we are displaying a given 

843 # namespace/root), tack on some dots to make it clear these names 

844 # require dotted paths to invoke. 

845 if ancestors or self.list_root: 

846 displayname = ".{}".format(displayname) 

847 aliases = [".{}".format(x) for x in aliases] 

848 # Nested? Indent, and add asterisks to default-tasks. 

849 if self.list_format == "nested": 

850 prefix = indent 

851 if is_default: 

852 displayname += "*" 

853 # Flat? Prefix names and aliases with ancestor names to get full 

854 # dotted path; and give default-tasks their collection name as the 

855 # first alias. 

856 if self.list_format == "flat": 

857 prefix = ancestor_path 

858 # Make sure leading dots are present for subcollections if 

859 # scoped display 

860 if prefix and self.list_root: 

861 prefix = "." + prefix 

862 aliases = [prefix + alias for alias in aliases] 

863 if is_default and ancestors: 

864 aliases.insert(0, prefix) 

865 # Generate full name and help columns and add to pairs. 

866 alias_str = " ({})".format(", ".join(aliases)) if aliases else "" 

867 full = prefix + displayname + alias_str 

868 pairs.append((full, helpline(task))) 

869 # Determine whether we're at max-depth or not 

870 truncate = self.list_depth and (len(ancestors) + 1) >= self.list_depth 

871 for name, subcoll in sorted(coll.collections.items()): 

872 displayname = name 

873 if ancestors or self.list_root: 

874 displayname = ".{}".format(displayname) 

875 if truncate: 

876 tallies = [ 

877 "{} {}".format(len(getattr(subcoll, attr)), attr) 

878 for attr in ("tasks", "collections") 

879 if getattr(subcoll, attr) 

880 ] 

881 displayname += " [{}]".format(", ".join(tallies)) 

882 if self.list_format == "nested": 

883 pairs.append((indent + displayname, helpline(subcoll))) 

884 elif self.list_format == "flat" and truncate: 

885 # NOTE: only adding coll-oriented pair if limiting by depth 

886 pairs.append((ancestor_path + displayname, helpline(subcoll))) 

887 # Recurse, if not already at max depth 

888 if not truncate: 

889 recursed_pairs = self._make_pairs( 

890 coll=subcoll, ancestors=ancestors + [name] 

891 ) 

892 pairs.extend(recursed_pairs) 

893 return pairs 

894 

895 def list_json(self) -> None: 

896 # Sanity: we can't cleanly honor the --list-depth argument without 

897 # changing the data schema or otherwise acting strangely; and it also 

898 # doesn't make a ton of sense to limit depth when the output is for a 

899 # script to handle. So we just refuse, for now. TODO: find better way 

900 if self.list_depth: 

901 raise Exit( 

902 "The --list-depth option is not supported with JSON format!" 

903 ) # noqa 

904 # TODO: consider using something more formal re: the format this emits, 

905 # eg json-schema or whatever. Would simplify the 

906 # relatively-concise-but-only-human docs that currently describe this. 

907 coll = self.scoped_collection 

908 data = coll.serialized() 

909 print(json.dumps(data))