1#!/usr/bin/env python3
2"""
3
4"""
5# Imports:
6from __future__ import annotations
7
8# ##-- stdlib imports
9import atexit# for @atexit.register
10import collections
11import contextlib
12import datetime
13import enum
14import faulthandler
15import functools as ftz
16import hashlib
17import itertools as itz
18import logging as logmod
19import os
20import pathlib as pl
21import re
22import sys
23import time
24import types
25from bdb import BdbQuit
26from copy import deepcopy
27from uuid import UUID, uuid1
28from weakref import ref
29
30# ##-- end stdlib imports
31
32# ##-- 3rd party imports
33import jgdv.cli
34import sh
35import stackprinter
36from jgdv import JGDVError, Mixin, Proto
37from jgdv.cli._interface import EMPTY_CMD, ParseReport_d
38from jgdv.cli.param_spec import ParamSpec, LiteralParam
39from jgdv.cli import ParamSpecMaker_m
40from jgdv.logging import JGDVLogConfig
41from jgdv.structs.chainguard import ChainGuard
42from jgdv.structs.chainguard._interface import ChainProxy_p
43from jgdv.util.plugins.selector import plugin_selector
44# ##-- end 3rd party imports
45
46# ##-- 1st party imports
47import doot
48import doot._interface as API # noqa: N812
49from doot.cmds._interface import AcceptsSubcmds_p
50import doot.errors as derrs
51
52# ##-- end 1st party imports
53
54# ##-- types
55# isort: off
56import abc
57import collections.abc
58from typing import TYPE_CHECKING, cast, assert_type, assert_never
59from typing import Generic, NewType
60# Protocols:
61from typing import Protocol, runtime_checkable
62# Typing Decorators:
63from typing import no_type_check, final, overload
64
65if TYPE_CHECKING:
66 from .loaders._interface import Loader_p
67 from ._interface import Main_p, Overlord_i
68 from typing import Final
69 from typing import ClassVar, Any, LiteralString
70 from typing import Never, Self, Literal
71 from typing import TypeGuard
72 from collections.abc import Iterable, Iterator, Callable, Generator
73 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
74
75 from logging import Logger
76 from jgdv import Maybe
77 from jgdv.cli import ParamSource_p, ParseMachine
78 from doot.errors import DootError
79
80 type DataSource = dict|ChainGuard
81 type LoaderDict = dict[str, Loader_p]
82
83##--|
84from doot.cmds._interface import Command_p
85from ._interface import Main_i
86# isort: on
87# ##-- end types
88
89##-- logging
90logging = logmod.getLogger(__name__)
91##-- end logging
92
93# Vars:
94env = os.environ
95DEFAULT_EMPTY_CMD : Final[list[str]] = ["--help"]
96DEFAULT_IMPLICIT_CMD : Final[list[str]] = ["run"]
97PROG_NAME : Final[str] = "doot"
98PARSER_FALLBACK : Final[str] = "doot.control.arg_parser_model:DootArgParserModel"
99PRE_COMMIT_K : Final[str] = "PRE_COMMIT"
100##--| controllers
101
[docs]
102class LoadingController:
103 """ mixin for triggering full loading """
104 type DM = Main_p
105
106 _version_template : str
107
[docs]
108 def load(self, obj:DM) -> None:
109 # Load and initialise the config:
110 doot.setup() # type: ignore[attr-defined]
111 # Then use it for everything else:
112 doot.load()
113 self.update_command_aliases(obj)
114 obj.setup_logging()
115
[docs]
116 def update_command_aliases(self, obj:DM) -> None:
117 """ Read settings.commands.* and register aliases
118
119 commands use doot.config.settings.commands.NAME,
120 and within that, 'aliases' gives a dict of {alias=[args]}
121
122 eg: commands.list.aliases.acts = ['--actions']
123 .. aliases 'doot acts'
124 .. to equiv of 'doot list --actions'
125
126 """
127 name : str
128 details : dict
129 registered : dict
130 logging.debug("Setting Command Aliases")
131 registered = {}
132 for name,details in doot.config.on_fail({}).settings.commands().items():
133 for alias, args in ChainGuard(details).on_fail({}, non_root=True).aliases().items():
134 logging.debug("- %s -> %s", name, alias)
135 registered[alias] = [name, *args]
136 else:
137 doot.cmd_aliases = ChainGuard(registered)
138 logging.debug("Finished Command Aliases")
139
[docs]
140class CLIController:
141 """ mixin for cli arg processing """
142 type DM = DootMain
143
[docs]
144 def parse_args(self, obj:DM, *, override:Maybe[list]=None) -> None:
145 """ use loaded cmd and tasks to parse sys.argv """
146 cmds : list
147 subcmds : list
148 unaliased_args : list[str]
149 implicits : dict[str,list[str]]
150 parser : ParseMachine
151 ##--|
152 parser = self._load_cli_parser(obj,
153 target=doot.config.on_fail("default").startup.loaders.parser())
154 cmds = list(doot.loaded_cmds.values())
155 subcmds = self._map_subcmd_constraints()
156 unaliased_args = self._unalias_raw_args(obj)
157 implicits = self._construct_implicits()
158
159 try:
160 cli_args = parser(unaliased_args,
161 prog=cast("ParamSource_p", obj),
162 cmds=cmds,
163 subs=subcmds,
164 implicits=implicits,
165 )
166 except jgdv.cli.errors.HeadParseError as err:
167 raise doot.errors.FrontendError("Doot Head Failed to Parse", err) from None
168 except jgdv.cli.errors.CmdParseError as err:
169 raise doot.errors.FrontendError("Unrecognised Command Called", err) from None
170 except jgdv.cli.errors.SubCmdParseError as err:
171 raise doot.errors.FrontendError("Unrecognised Task Called", err.args[1]) from None
172 except jgdv.cli.errors.ArgParseError as err:
173 raise doot.errors.FrontendError("Parsing arguments for command/task failed", err) from None
174 except jgdv.cli.ParseError as err:
175 raise doot.errors.ParseError("Failed to Parse provided cli args", err) from None
176 else:
177 match cli_args:
178 case ParseReport_d()|dict() as parsed_args:
179 doot.update_cmd_args(parsed_args, override=bool(override)) # type: ignore[attr-defined]
180 case x:
181 raise TypeError(type(x))
182
[docs]
183 def _load_cli_parser(self, obj:DM, *, target:str="default") -> ParseMachine:
184 match plugin_selector(doot.loaded_plugins.on_fail([], list).parser(),
185 target=target,
186 fallback=PARSER_FALLBACK):
187 case None:
188 parser_model = None
189 case type() as ctor:
190 parser_model = ctor()
191 case x:
192 raise TypeError(type(x))
193
194 match parser_model:
195 case None:
196 from .arg_parser_model import DootArgParserModel # noqa: PLC0415
197 return jgdv.cli.ParseMachine(DootArgParserModel())
198 case jgdv.cli.ArgParserModel_p() as p:
199 return jgdv.cli.ParseMachine(parser=p)
200 case _:
201 raise TypeError("Improper parser model specified", parser_model)
202
[docs]
203 def _unalias_raw_args(self, obj:DM) -> list[str]:
204 """ replaces aliases with their full command args.
205
206 Just a simple, literal, find and replace
207 """
208 raw : list[str]
209 result = []
210 sep = doot.constants.patterns.TASK_PARSE_SEP # type: ignore[attr-defined]
211
212 match obj.raw_args:
213 case []:
214 raise ValueError("No args were found")
215 case [x]:
216 empty_cmd = doot.config.on_fail(["--help"]).startup.empty_cmd()
217 raw = [x, *empty_cmd]
218 case [*xs]:
219 raw = xs
220
221 for i, x in enumerate(raw):
222 match doot.cmd_aliases.on_fail(None)[x]():
223 case None:
224 # cmd name is not an alias
225 result.append(x)
226 case [name, *_] as args if name in doot.loaded_cmds:
227 # is an alias
228 logging.debug("Using Alias: %s -> %s", x, args)
229 result += args
230 case x:
231 raise TypeError(type(x))
232 else:
233 return result
234
[docs]
235 def _construct_implicits(self) -> dict[str, list[str]]:
236 result : dict = {}
237 match doot.config.on_fail(DEFAULT_IMPLICIT_CMD, list).startup.implicit_cmd():
238 case [x, *_] as xs:
239 result[x] = xs
240 case []:
241 pass
242 case x:
243 raise TypeError(type(x))
244 return result
245
[docs]
246 def _map_subcmd_constraints(self) -> list[tuple[tuple[str, ...], ParamSource_p]]:
247 subcmd_handlers = tuple(x for x,y in doot.loaded_cmds.items() if isinstance(y, AcceptsSubcmds_p))
248 subcmds = [(subcmd_handlers, x) for x in doot.loaded_tasks.values()]
249 return subcmds
250
[docs]
251class CmdController:
252 """ mixin for actually running a command """
253 type DM = DootMain
254
[docs]
255 def prepare(self, obj:DM) -> None:
256 pass
257
[docs]
258 def run_cmds(self, obj:DM) -> None:
259 name : str
260 calls : list[dict]
261 try:
262 for name, calls in doot.args.cmds.items(): # type: ignore[attr-defined]
263 cmd = self.get_cmd_instance(obj, cmd=name)
264 for idx in range(len(calls)):
265 obj.result_code = self.run_cmd(idx=idx, cmd=cmd)
266 else:
267 pass
268 else:
269 pass
270
271 except doot.errors.DootError as err:
272 obj._errored = err
273 raise
274 else:
275 pass
276
[docs]
277 def get_cmd_instance(self, obj:DM, *, cmd:str) -> Command_p:
278 """ Uses the full command name to get the instance of the command """
279 x : Any
280 target : str
281 ##--|
282 logging.debug("Initial Retrieval attempt: %s", cmd)
283 match doot.loaded_cmds.get(cmd, None):
284 case Command_p() as x:
285 return x
286 case x:
287 raise TypeError(type(x), x)
288
[docs]
289 def run_cmd(self, *, idx:int, cmd:Command_p) -> int:
290 """
291 The method run to trigger a doot workflow
292
293 """
294 match cmd:
295 case Command_p() as cmd:
296 pass
297 case x:
298 return API.ExitCodes.BAD_CMD
299
300 # Do the cmd
301 logging.info("Doot Calling Cmd: %s", cmd.name)
302 cmd(idx=idx, tasks=doot.loaded_tasks, plugins=doot.loaded_plugins)
303 return API.ExitCodes.SUCCESS
304
[docs]
305class ShutdownController:
306 """ mixin for cleaning up on and shutting down doot """
307 type DM = DootMain
308
[docs]
309 def prepare(self, obj:DM) -> None:
310 self.install_handler(obj)
311
[docs]
312 def shutdown(self, obj:DM) -> None:
313 """ Doot has finished, report on what was done and how doot finished"""
314 logging.info("Shutting Down Doot")
315 match obj.current_cmd:
316 case None:
317 pass
318 case Command_p() as cmd:
319 cmd.shutdown(doot.loaded_tasks, doot.loaded_plugins, errored=obj._errored)
320
321 self.record_defaulted_config_values()
322
323 doot.report.gen.line()
324 match obj._errored:
325 case doot.errors.DootError() as err:
326 logging.exception("fail")
327 case Exception() as err:
328 raise err
329 case None:
330 pass
331
332 doot.report.summary.summarise()
333
[docs]
334 def install_handler(self, obj:DM) -> None:
335 """ Install an exit handler """
336 report_val : str
337 match doot.config.on_fail(None).shutdown.notify.exit():
338 case None: # No config value, use default
339 report_val = "Dooted"
340 case False:
341 return
342 case str() as x:
343 report_val = x
344
345 def goodbye(*args, **kwargs) -> None: # noqa: ARG001, ANN002, ANN003
346 doot.report.gen.line(report_val)
347
348 atexit.register(goodbye)
349
[docs]
350 def announce_exit(self, message:str) -> None:
351 """ triggers speech synthesis on exiting doot """
352 if not doot.config.on_fail(False).shutdown.notify.say_on_exit(): # noqa: FBT003
353 return
354
355 match sys.platform:
356 case _ if PRE_COMMIT_K in env:
357 return
358 case "linux":
359 sh.espeak(message) # type: ignore[attr-defined]
360 case "darwin":
361 sh.say("-v", "Moira", "-r", "50", message) # type: ignore[attr-defined]
362
[docs]
363 def record_defaulted_config_values(self) -> None:
364 if not doot.config.on_fail(False).shutdown.write_defaulted_values(): # noqa: FBT003
365 return
366
367 defaulted_file : str = doot.config.on_fail("{logs}/.doot_defaults.toml", str).shutdown.defaulted_values.path()
368 expanded_path : pl.Path = doot.locs[defaulted_file]
369 if not expanded_path.parent.exists():
370 logging.error("Couldn't log defaulted config values to: %s", expanded_path)
371 return
372
373 defaulted_toml = ChainGuard.report_defaulted() # type: ignore[attr-defined]
374 with pl.Path(expanded_path).open('w') as f:
375 f.write("# default values used:\n")
376 f.write("\n".join(defaulted_toml) + "\n\n")
377
[docs]
378class ErrorHandlers:
379 """ Mixin for handling different errors of doot """
380 type DM = DootMain
381
[docs]
382 def discriminate_exit(self, obj:DootMain, err:Exception) -> int:
383 result : int
384 match err:
385 case derrs.EarlyExit() | derrs.Interrupt() | BdbQuit():
386 result = self._early_exit(err)
387 case derrs.MissingConfigError():
388 result = self._missing_config_exit(obj, err)
389 case derrs.ConfigError():
390 result = self._config_error_exit(err)
391 case derrs.TaskFailed() | derrs.TaskError():
392 result = self._task_failed_exit(err)
393 case derrs.StateError():
394 result = self._bad_state_exit(err)
395 case derrs.StructLoadError():
396 result = self._bad_struct_exit(err)
397 case derrs.TrackingError():
398 result = self._tracking_exit(err)
399 case derrs.BackendError():
400 result = self._backend_exit(err)
401 case derrs.FrontendError():
402 result = self._frontend_exit(err)
403 case derrs.DootError():
404 result = self._misc_doot_exit(err)
405 case NotImplementedError():
406 result = self._not_implemented_exit(err)
407 case _:
408 result = self.python_exit(err)
409 ##--|
410 return result
411
[docs]
412 def _early_exit(self, err:derrs.EarlyExit|derrs.Interrupt|BdbQuit) -> int: # noqa: ARG002
413 logging.warn("Early Exit Triggered")
414 return API.ExitCodes.EARLY
415
[docs]
416 def _missing_config_exit(self, obj:DootMain, err:derrs.MissingConfigError) -> int:
417 load_targets : list
418 base_target : pl.Path
419 ##--|
420 match obj.raw_args:
421 case [*_, "stub", "--config"]:
422 from doot.cmds.stub_cmd import StubCmd # noqa: PLC0415
423 stubber = StubCmd()
424 stubber._stub_doot_toml() # type: ignore[attr-defined]
425 return 0
426 case _:
427 pass
428
429 load_targets = doot.constants.on_fail(["doot.toml"]).paths.DEFAULT_LOAD_TARGETS()
430 base_target = pl.Path(load_targets[0])
431 # Handle missing files
432 if base_target.exists():
433 logging.error("[%s] : Base Config Target exists but it contains no valid config: %s",
434 type(err).__name__, base_target)
435 else:
436 logging.warn("[%s] : No toml config data found, create a doot.toml by calling `doot stub --config`",
437 type(err).__name__)
438
439 return API.ExitCodes.MISSING_CONFIG
440
[docs]
441 def _config_error_exit(self, err:derrs.ConfigError) -> int:
442 logging.warn("[%s] : Config Error: %s", type(err).__name__, err)
443 return API.ExitCodes.BAD_CONFIG
444
[docs]
445 def _task_failed_exit(self, err:derrs.TaskError) -> int:
446 logging.error("[%s] : Task Error : %s", type(err).__name__, err, exc_info=err)
447 if hasattr(err, "task_source"):
448 logging.error("[%s] : Task Source: %s", type(err).__name__, err.task_source)
449 return API.ExitCodes.TASK_FAIL
450
[docs]
451 def _bad_state_exit(self, err:derrs.StateError) -> int:
452 logging.error("[%s] : State Error: %s", type(err).__name__, err.args)
453 return API.ExitCodes.BAD_STATE
454
[docs]
455 def _bad_struct_exit(self, err:derrs.StructLoadError) -> int:
456 match err.args:
457 case [str() as msg, dict() as errs]:
458 logging.error("[%s] : Struct Load Errors : %s", type(err).__name__, msg)
459 logging.error("")
460 for x,y in errs.items():
461 logging.error("---- File: %s", x)
462 for val in y:
463 logging.error("- %s", val)
464 else:
465 logging.error("")
466 case _:
467 logging.error("[%s] : Struct Load Error: %s", type(err).__name__, err, exc_info=err)
468
469 return API.ExitCodes.BAD_STRUCT
470
[docs]
471 def _tracking_exit(self, err:derrs.TrackingError) -> int:
472 logging.error("[%s] : Tracking Failure: %s", type(err).__name__, err.args)
473 return API.ExitCodes.TRACKING_FAIL
474
[docs]
475 def _backend_exit(self, err:derrs.BackendError) -> int:
476 logging.error("[%s] : Backend Error: %s", type(err).__name__, err.args, exc_info=err)
477 return API.ExitCodes.BACKEND_FAIL
478
[docs]
479 def _frontend_exit(self, err:derrs.FrontendError) -> int:
480 logging.error("[%s] : %s", type(err).__name__, " : ".join(err.args))
481 return API.ExitCodes.FRONTEND_FAIL
482
[docs]
483 def _misc_doot_exit(self, err:derrs.DootError) -> int:
484 logging.error("[%s] : %s", type(err).__name__, err.args, exc_info=err)
485 return API.ExitCodes.DOOT_FAIL
486
[docs]
487 def _not_implemented_exit(self, err:NotImplementedError) -> int:
488 logging.error("[%s] : Not Implemented: %s", type(err).__name__, err.args, exc_info=err)
489 return API.ExitCodes.NOT_IMPLEMENTED
490
[docs]
491 def python_exit(self, err:Exception) -> int:
492 lasterr : pl.Path
493 logging.error("[%s] : Python Error:", type(err).__name__, exc_info=err)
494 lasterr = pl.Path(API.LASTERR).resolve()
495 lasterr.write_text(stackprinter.format())
496 logging.error(f"[{type(err).__name__}] : Python Error, full stacktrace written to {lasterr}", exc_info=None)
497 return API.ExitCodes.PYTHON_FAIL
498
499##--|
500
[docs]
501@Proto(Main_i)
502class DootMain(ParamSpecMaker_m):
503 """ doot.main and the associated exit handlers
504
505 Error's if doot hasn't got an overlord (aliased as the doot package)
506
507 loads values from the overlord config,
508 sets up runtime plugin system
509
510 """
511 _loading : ClassVar[LoadingController] = LoadingController()
512 _cli : ClassVar[CLIController] = CLIController()
513 _cmd : ClassVar[CmdController] = CmdController()
514 _shutdown : ClassVar[ShutdownController] = ShutdownController()
515 _err : ClassVar[ErrorHandlers] = ErrorHandlers()
516
517 ##--|
518 result_code : int
519 bin_name : str
520 prog_name : str
521 current_cmd : Maybe[Command_p]
522 _errored : Maybe[Exception]
523 _help_txt = tuple(["A Toml Specified Task Runner"])
524
525 def __init__(self, *, cli_args:Maybe[list]=None) -> None:
526 match cli_args:
527 case None:
528 self.raw_args = sys.argv[:]
529 case list() as vals:
530 self.raw_args = vals
531 case x:
532 raise TypeError(type(x))
533
534 ##--|
535 self.result_code = API.ExitCodes.INITIAL
536 self.bin_name = pl.Path(self.raw_args[0]).name
537 self.prog_name = "doot"
538 self.current_cmd = None
539 self.parser = None
540 self._errored = None
541 self.log_config = JGDVLogConfig()
542
[docs]
543 @property
544 def name(self) -> str:
545 return PROG_NAME
546
[docs]
547 def param_specs(self) -> list[ParamSpec]:
548 """ The cli parameters of the main doot program. """
549 return [
550 # TODO may need to control the sort of this literal
551 LiteralParam(name=self.prog_name, desc="The Program Name"),
552 self.build_param(name="--version" , type=bool, desc="Print the version number"),
553 self.build_param(name="--help" , type=bool, desc="Print this help"),
554
555 self.build_param(name="--verbose" , type=bool, desc="Increase Verbosity"),
556 self.build_param(name="--debug", type=bool, desc="Activate breakpoints"),
557 ]
558
[docs]
559 def help(self) -> str:
560 help_lines = ["", f"Doot v{doot.__version__}", ""]
561 help_lines += self._help_txt
562
563 params = self.param_specs()
564 if bool(params):
565 help_lines += ["", "Params:"]
566 help_lines += (x.help_str() for x in self.param_specs())
567
568 help_lines.append("")
569 help_lines.append("Commands: ")
570 help_lines += sorted(x.helpline for x in doot.loaded_cmds.values())
571
572 return "\n".join(help_lines)
573
[docs]
574 def setup_logging(self) -> None:
575 self.log_config.setup(doot.config)
576 doot.load_reporter()
577
[docs]
578 def handle_cli_args(self) -> Maybe[int]:
579 """ Overlord specific cli arg responses.
580 Modifies:
581 - verbosity,
582 - print version
583 - header suppression
584 - help printing
585 - debugger entry
586
587 return an int to give an override result code
588 """
589 _version_template : str
590 ##--|
591 _version_template = doot.constants.printer.version_template # type: ignore[attr-defined]
592 if doot.args.on_fail(False).prog.args.verbose(): # noqa: FBT4
593 logging.info("Switching to Verbose Output")
594 self.log_config.set_level("DEBUG")
595
596 if doot.args.on_fail(False).prog.args.version(): # noqa: FBT003
597 logging.info(_version_template, API.__version__)
598 return API.ExitCodes.SUCCESS
599
600 if not doot.args.on_fail(False).prog.args.suppress_header(): # noqa: FBT003
601 doot.report.gen.header()
602
603 if not bool(doot.args.on_fail({}).cmds()) and doot.args.on_fail(False).help(): # noqa: FBT003
604 helptxt = self.help()
605 logging.warning(helptxt)
606 return API.ExitCodes.SUCCESS
607
608 if doot.args.on_fail(False).prog.args.debug(): # noqa: FBT003
609 logging.info("Pausing for debugging")
610 breakpoint()
611 pass
612
613 return None
614
615 def __call__(self) -> None:
616 """ The Main Doot CLI Program.
617 Loads data and plugins before starting the requested command.
618
619 Catches: doot errors, then NotImplementedError, then Exception
620
621 has a 'finally' block to call sys.exit
622 """
623 x : Any
624 try:
625 self._loading.load(self)
626 self._cli.parse_args(self)
627 match self.handle_cli_args():
628 case None:
629 pass
630 case int() as x:
631 self.result_code = x
632 return
633
634 self._shutdown.prepare(self)
635 self._cmd.run_cmds(self)
636 except (derrs.DootError, BdbQuit, NotImplementedError) as err:
637 self.result_code = self._err.discriminate_exit(self, err)
638 except Exception as err: # noqa: BLE001
639 self.result_code = self._err.python_exit(err)
640 finally:
641 self._shutdown.shutdown(self)
642 sys.exit(self.result_code)