1#!/usr/bin/env python3
2"""
3Utility classes for building tasks with a bit of structure
4"""
5# Imports:
6from __future__ import annotations
7
8# ##-- stdlib imports
9import datetime
10import enum
11import functools as ftz
12import itertools as itz
13import logging as logmod
14import pathlib as pl
15import re
16import time
17from uuid import UUID, uuid1
18
19# ##-- end stdlib imports
20
21# ##-- 3rd party imports
22from jgdv import Mixin, Proto
23from jgdv.structs.chainguard import ChainGuard
24from jgdv.structs.strang import CodeReference
25
26# ##-- end 3rd party imports
27
28# ##-- 1st party imports
29import doot
30import doot.errors
31
32# ##-- end 1st party imports
33
34# ##-| Local
35from ._interface import Job_p, Task_p, TaskMeta_e, TaskSpec_i, TaskName_p
36from .structs.task_name import TaskName
37from .structs.task_spec import TaskSpec
38from .task import DootTask
39
40# # End of Imports.
41
42# ##-- types
43# isort: off
44import abc
45import collections.abc
46from typing import TYPE_CHECKING, cast, assert_type, assert_never
47from typing import Generic, NewType
48# Protocols:
49from typing import Protocol, runtime_checkable
50# Typing Decorators:
51from typing import no_type_check, final, override, overload
52
53if TYPE_CHECKING:
54 from jgdv import Maybe
55 from typing import Final
56 from typing import ClassVar, Any, LiteralString
57 from typing import Never, Self, Literal
58 from typing import TypeGuard
59 from collections.abc import Iterable, Iterator, Callable, Generator
60 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
61 from doot.cmds.structs.task_stub import TaskStub
62
63# isort: on
64# ##-- end types
65
66##-- logging
67logging = logmod.getLogger(__name__)
68##-- end logging
69
70SUBTASKED_HEAD = doot.constants.patterns.SUBTASKED_HEAD # type: ignore[attr-defined]
71
[docs]
72class _JobStubbing_m:
73
[docs]
74 @classmethod
75 def stub_class(cls, stub:TaskStub) -> TaskStub:
76 # Come first
77 stub['active_when'].priority = -90
78 stub['required_for'].priority = -90
79 stub['depends_on'].priority = -100
80
81 stub['head_task'].set(type="taskname", default="", prefix="# ", priority=100)
82 stub['queue_behaviour'].default = "default"
83 stub['queue_behaviour'].comment = "default | auto | reactive"
84 return stub
85
[docs]
86 def stub_instance(self, stub:TaskStub) -> TaskStub:
87 stub = self.__class__.stub_class(stub)
88 stub['name'].default = self.name.de_uniq() # type: ignore[attr-defined]
89 if bool(self.doc): # type: ignore[attr-defined]
90 stub['doc'].default = [f"\"{x}\"" for x in self.doc] # type: ignore[attr-defined]
91 return stub
92
[docs]
93@Proto(Job_p, check=True)
94@Mixin(_JobStubbing_m)
95class DootJob(DootTask):
96 """ Util Class for building single tasks
97 wraps with setup and teardown tasks,
98 manages cleaning,
99 and holds state
100
101 """
102 _help = tuple(["A Basic Task Constructor"])
103 _default_flags : ClassVar = {TaskMeta_e.JOB}
104 version : str = "0.1"
105
[docs]
106 @classmethod
107 def class_help(cls) -> list[str]:
108 """ Job *class* help. """
109 version = getattr(cls, "_version", "0.1")
110 help_lines = [f"Job : {cls.__qualname__} v{version} ({cls.__module__}:{cls.__qualname__})", ""]
111
112 mro = " -> ".join(x.__name__ for x in cls.mro())
113 help_lines.append(f"Job MRO: {mro}")
114 help_lines.append("")
115 help_lines += cls._help
116
117 match getattr(cls, "param_specs", None):
118 case None:
119 params = []
120 case x if callable(x):
121 params = x.param_specs()
122
123 if not bool(params):
124 return help_lines
125
126 help_lines += ["", "Params:"]
127 for param in params:
128 if (param_help:=str(param)):
129 help_lines.append(param_help)
130
131 return help_lines
132
133 def __init__(self, spec:TaskSpec_i):
134 assert(spec is not None), "Spec is empty"
135 super().__init__(spec)
136
[docs]
137 def default_task(self, name:Maybe[str|TaskName_p], extra:Maybe[dict|ChainGuard]) -> TaskSpec_i:
138 task_name = None
139 match name:
140 case None:
141 task_name = self.name.push(SUBTASKED_HEAD)
142 case str():
143 task_name = self.name.push(name)
144 case TaskName():
145 task_name = name
146 case _:
147 raise doot.errors.StructError("Bad value used to make a subtask in %s : %s", self.name.de_uniq(), name)
148
149 assert(task_name is not None)
150 return TaskSpec(name=task_name, extra=ChainGuard(extra))
151
[docs]
152 def is_stale(self, task:Task_p) -> bool: # noqa: ARG002
153 return False
154
[docs]
155 def specialize_task(self, task:Task_p) -> Task_p:
156 return task
157
[docs]
158 def expand_job(self) -> list:
159 return []