1#!/usr/bin/env python3
2"""
3The command which provides the user a listing of... things.
4
5Each mixin does the work of creating the list, and provides its own
6set of cli args.
7The actual ListCmd joins them all and calls the necessary method,
8passing the result to the generic command _print_text method
9"""
10# ruff: noqa: ANN001
11# Imports:
12from __future__ import annotations
13
14# ##-- stdlib imports
15import datetime
16import enum
17import functools as ftz
18import itertools as itz
19import logging as logmod
20import pathlib as pl
21import re
22import time
23import types
24import typing
25from collections import defaultdict
26from copy import deepcopy
27from uuid import UUID, uuid1
28from weakref import ref
29
30# ##-- end stdlib imports
31
32# ##-- 3rd party imports
33from jgdv import Proto, Mixin
34from jgdv.structs.strang import Strang
35from jgdv.logging import _interface as LogAPI # noqa: N812
36
37# ##-- end 3rd party imports
38
39# ##-- 1st party imports
40import doot
41import doot.errors
42from doot.workflow._interface import TaskMeta_e
43from doot.workflow import TaskName
44from ._base import BaseCommand
45from ._interface import Command_p
46
47# ##-- end 1st party imports
48
49# ##-- types
50# isort: off
51import abc
52import collections.abc
53from typing import TYPE_CHECKING, cast, assert_type, assert_never
54from typing import Generic, NewType
55# Protocols:
56from typing import Protocol, runtime_checkable
57# Typing Decorators:
58from typing import no_type_check, final, override, overload
59
60if TYPE_CHECKING:
61 from jgdv import Maybe, Rx
62 from typing import Final
63 from typing import ClassVar, Any, LiteralString
64 from typing import Never, Self, Literal
65 from typing import TypeGuard
66 from collections.abc import Iterable, Iterator, Callable, Generator
67 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
68
69 from jgdv.cli import ParamSpec_p
70 from jgdv.cli.param_spec import ParamSpec
71 from jgdv.structs.chainguard import ChainGuard
72
73 type ListVal = Maybe[str|tuple[str, dict]]
74
75# isort: on
76# ##-- end types
77
78##-- logging
79logging = logmod.getLogger(__name__)
80##-- end logging
81
82GROUP_INDENT : Final[str] = " "*4
83INDENT : Final[str] = " "*8
84FMT_STR : Final[str] = doot.config.on_fail("{indent}{val}").settings.command.list.fmt()
85hide_names : Final[list[str]] = doot.config.on_fail([]).settings.commands.list.hide()
86hide_re : Final[Rx] = re.compile("^({})".format("|".join(hide_names)))
87
88##--|
89
[docs]
90class _DagLister_m:
91 build_param : Callable
92
[docs]
93 def param_specs(self) -> list:
94 return [
95 *super().param_specs(), # type: ignore[misc]
96 self.build_param(name="--dag",
97 _short="D",
98 type=bool,
99 default=False,
100 desc="Output a DOT compatible graph of tasks"),
101 ]
102
[docs]
103class _TaskLister_m:
104 """
105 TODO: colour jobs
106 """
107 build_param : Callable
108
[docs]
109 def param_specs(self) -> list:
110 return [
111 *super().param_specs(), # type: ignore[misc]
112 self.build_param(name="-tasks",
113 type=bool,
114 default=True,
115 desc="List all loaded tasks, by group"),
116 # Task Listing Parameters
117 self.build_param(name="--group-by=",
118 default="group",
119 desc="How to group listed tasks"),
120 self.build_param(name="+dependencies", default=False, desc="List task dependencies"),
121 self.build_param(name="+internal", default=False, desc="Include internal tasks (ie: prefixed with an underscore)"),
122 self.build_param(name="+docstr", default=False),
123 self.build_param(name="+params", default=False),
124 ]
125
[docs]
126 def _list_tasks(self, idx:int, tasks) -> list[ListVal]:
127 logging.info("---- Listing tasks")
128
129 result : list[ListVal] = []
130 result.append(("Registered Tasks/Jobs:", {"colour":"cyan"}))
131 if not bool(tasks):
132 result.append(("!! No Tasks Defined", {"colour":"cyan"}))
133 return result
134
135 max_key = len(max(tasks.keys(), key=len, default="def"))
136 fmt_strs, base_vars = self._build_format_strings(idx)
137 data : list[dict] = []
138
139 logging.info("-- Collecting Tasks")
140 for spec in tasks.values():
141 data.append(
142 base_vars | {
143 "indent" : " "*(1 + len(GROUP_INDENT) + len(spec.name[0,:]) + 2),
144 "internal" : TaskName.Marks.hide in spec.name,
145 "disabled" : TaskMeta_e.DISABLED in spec.meta,
146 "group" : spec.name[0,:],
147 "val" : spec.name[1,:],
148 "full" : spec.name,
149 "docstr" : (spec.doc[0] if bool(spec.doc) else "")[:60],
150 "source" : (spec.sources[0] if bool(spec.sources) else "(No Source)"),
151 },
152 )
153
154 data = self._filter_tasks(idx, data)
155 grouped = self._group_tasks(data)
156
157 logging.info("-- Formatting")
158 for group, items in grouped.items():
159 result.append((f"*{GROUP_INDENT}{group}::", {"colour":"magenta"}))
160 for task in items:
161 result.append(" ".join(x.format_map(task) for x in fmt_strs))
162
163 else:
164 return result
165
195
[docs]
196 def _filter_tasks(self, idx:int, data) -> list[dict]:
197 logging.info("-- Filtering: %s", len(data))
198 show_internal : bool = doot.args.on_fail(False).cmd.args.internal() # noqa: FBT003
199 no_hide_names = bool(hide_names)
200 match doot.args.on_fail(None).cmds[self.name][idx].args.pattern().lower():
201 case None | "":
202 pattern = None
203 case str() as x:
204 pattern = re.compile(x, flags=re.IGNORECASE)
205
206 def _filter_fn(item:dict) -> bool:
207 return all([not item['disabled'],
208 show_internal or not item['internal'],
209 pattern is None or pattern.match(item['full']),
210 no_hide_names or hide_re.search(item['full']),
211 ])
212
213 return list(filter(_filter_fn, data))
214
[docs]
215 def _group_tasks(self, data:list[dict]) -> dict[str,list]:
216 logging.info("-- Grouping: %s", len(data))
217 result = defaultdict(list)
218
219 match doot.args.on_fail(None).cmd.args.group_by():
220 case None | "group":
221
222 def _group_fn(item) -> str:
223 return item['group']
224 case "source":
225
226 def _group_fn(item) -> str:
227 return item['source']
228 case x:
229 raise ValueError("Unknown group-by arg", x)
230
231 for item in data:
232 result[_group_fn(item)].append(item)
233
234 return result
235
[docs]
236class _LocationLister_m:
237 build_param : Callable
238
[docs]
239 def param_specs(self) -> list:
240 return [
241 *super().param_specs(), # type: ignore[misc]
242 self.build_param(name="-locs",
243 type=bool,
244 default=False,
245 desc="List all Loaded Locations"),
246 ]
247
[docs]
248 def _list_locations(self, idx) -> list[ListVal]:
249 logging.info("---- Listing Defined Locations")
250 result : list[ListVal] = []
251 result.append("Defined Locations: ")
252
253 assert(doot.locs.Current is not None)
254 for x in sorted(doot.locs.Current):
255 loc = doot.locs.Current.get(x)
256 result.append(f"-- {x:<25} : {loc} ")
257 else:
258 return result
259
[docs]
260class _LoggerLister_m:
261 build_param : Callable
262
[docs]
263 def param_specs(self) -> list:
264 return [
265 *super().param_specs(), # type: ignore[misc]
266 self.build_param(name="-loggers",
267 type=bool,
268 default=False,
269 desc="List All Print Points"),
270 ]
271
[docs]
272 def _list_loggers(self, idx) -> list[ListVal]:
273 logging.info("---- Listing Logging/Printing info")
274
275 result : list[ListVal] = []
276
277 result.append("--- Primary Loggers:")
278 result.append("- doot.report ( target= ) : For user-facing output")
279 result.append("- stream ( target= )")
280 result.append("- file ( target= filename_fmt=%str ) ")
281
282 result.append(None)
283 result.append("--- Logging Targets: (Where a logger outputs to)")
284 result += [ f"- {x}" for x in LogAPI.TARGETS ]
285
286 result.append(None)
287 result.append("--- Notes: ")
288 result.append("Format is the {} form of log formatting")
289 result.append("Available variables are found here:")
290 result.append("https://docs.python.org/3/library/logging.html#logrecord-attributes")
291 result.append(None)
292 return result
293
[docs]
294class _FlagLister_m:
295 build_param : Callable
296
[docs]
297 def param_specs(self) -> list:
298 return [
299 *super().param_specs(), # type: ignore[misc]
300 self.build_param(name="-flags",
301 type=bool,
302 default=False,
303 desc="List Task Meta"),
304 ]
305
[docs]
306 def _list_flags(self, idx) -> list[ListVal]:
307 logging.info("---- Listing Task Flags")
308 result : list[ListVal] = []
309 result.append("Task Flags: ")
310 for x in sorted(TaskMeta_e):
311 result.append(f"-- {x}")
312 else:
313 return result
314
[docs]
315class _ActionLister_m:
316 build_param : Callable
317
[docs]
318 def param_specs(self) -> list:
319 return [
320 *super().param_specs(), # type: ignore[misc]
321 self.build_param(name="-actions",
322 type=bool,
323 default=False,
324 desc="List All Known Actions"),
325 ]
326
[docs]
327 def _list_actions(self, idx, plugins) -> list[ListVal]:
328 logging.info("---- Listing Available Actions")
329 result : list[ListVal] = []
330 result.append("Available Actions:")
331 largest = len(max((x.name for x in plugins.action), key=len, default=""))
332 for action in sorted(plugins.action, key=lambda x: x.name):
333 result.append(f"-- {action.name:<25} : {action.value}")
334 else:
335 result.append(None)
336 result.append("- For Custom Python Actions, implement the following in the .tasks directory")
337 result.append("def custom_action(spec:ActionSpec, task_state:dict) -> Maybe[bool|dict]:...")
338 return result
339
[docs]
340class _PluginLister_m:
341 build_param : Callable
342
[docs]
343 def param_specs(self) -> list:
344 return [
345 *super().param_specs(), # type: ignore[misc]
346 self.build_param(name="-plugins",
347 type=bool,
348 default=False,
349 desc="List All Known Plugins"),
350 ]
351
[docs]
352 def _list_plugins(self, idx, plugins) -> list[ListVal]:
353 logging.info("---- Listing Plugins")
354 result : list[ListVal] = []
355 result.append(("Defined Plugins by Group:", {"colour":"cyan"}))
356 max_key : int = len(max(plugins.keys(), key=len, default=""))
357 fmt_str : str = f"{INDENT}%-{max_key}s :: %-25s"
358 groups : dict = defaultdict(list)
359 for group_str, specs in plugins.items():
360 groups[group_str] += [(spec.name, spec.value) for spec in specs]
361
362 for group, items in groups.items():
363 if not bool(items):
364 continue
365 result.append((f"* {group}::", {"colour":"magenta"}))
366 for plugin in items:
367 result.append(fmt_str % plugin)
368
369 result.append(None)
370 return result
371
372##--|
373
[docs]
374@Mixin(_TaskLister_m, _LocationLister_m, _LoggerLister_m)
375@Mixin(_FlagLister_m, _ActionLister_m, _PluginLister_m)
376class _Listings_m:
377 pass
378
[docs]
379@Proto(Command_p)
380@Mixin(_Listings_m, None, allow_inheritance=True)
381class ListCmd(BaseCommand):
382 build_param : Callable
383 _help : ClassVar = tuple([
384 "A simple command to list all loaded task heads.",
385 "Set settings.commands.list.hide with a list of regexs to ignore",
386 ])
387
388 def __init__(self, *args, **kwargs) -> None:
389 super().__init__(*args, **kwargs, name="list")
390
[docs]
391 @override
392 def param_specs(self) -> list[ParamSpec_p]:
393 params = [
394 *super().param_specs(),
395 self.build_param(name="<0>pattern", type=str, default="", desc="Filter the listing to only values passing this regex"),
396 ]
397 return params
398
399 def __call__(self, *, idx:int, tasks:ChainGuard, plugins:ChainGuard):
400 """List task generators"""
401 assert(isinstance(tasks, dict))
402 assert(isinstance(plugins, dict))
403 logging.debug("Starting to List Jobs/Tasks")
404 result : list[Maybe[str]] = []
405 match dict(doot.args.on_fail({}).cmds[self.name][idx].args()):
406 case {"flags":True}:
407 result = self._list_flags(idx) # type: ignore[attr-defined]
408 case {"loggers":True}:
409 result = self._list_loggers(idx) # type: ignore[attr-defined]
410 case {"actions":True}:
411 result = self._list_actions(idx, plugins) # type: ignore[attr-defined]
412 case {"plugins":True}:
413 result = self._list_plugins(idx, plugins) # type: ignore[attr-defined]
414 case {"locs": True}:
415 result = self._list_locations(idx) # type: ignore[attr-defined]
416 case {"tasks": True}:
417 result = self._list_tasks(idx, tasks) # type: ignore[attr-defined]
418 case _:
419 raise doot.errors.CommandError("Bad args passed in", dict(doot.args))
420
421 self._print_text(result)