Source code for doot.workflow.actions.util.decorators

  1#!/usr/bin/env python3
  2"""
  3Action Decorators for metadata.
  4
  5"""
  6
  7# Imports:
  8from __future__ import annotations
  9
 10# ##-- stdlib imports
 11import datetime
 12import enum
 13import functools as ftz
 14import itertools as itz
 15import logging as logmod
 16import pathlib as pl
 17import re
 18import time
 19import types
 20import weakref
 21from uuid import UUID, uuid1
 22
 23# ##-- end stdlib imports
 24
 25# ##-- 3rd party imports
 26from jgdv.decorators import MetaDec, IdempotentDec
 27
 28# ##-- end 3rd party imports
 29
 30# ##-- 1st party imports
 31import doot
 32import doot.errors
 33
 34# ##-- end 1st party imports
 35
 36# ##-- types
 37# isort: off
 38import abc
 39import collections.abc
 40from typing import TYPE_CHECKING, cast, assert_type, assert_never
 41from typing import Generic, NewType
 42# Protocols:
 43from typing import Protocol, runtime_checkable
 44# Typing Decorators:
 45from typing import no_type_check, final, override, overload
 46
 47if TYPE_CHECKING:
 48    from jgdv import Maybe
 49    from typing import Final
 50    from typing import ClassVar, Any, LiteralString
 51    from typing import Never, Self, Literal
 52    from typing import TypeGuard
 53    from collections.abc import Iterable, Iterator, Callable, Generator
 54    from collections.abc import Sequence, Mapping, MutableMapping, Hashable
 55
 56##--|
 57from jgdv._abstract.protocols.general import SpecStruct_p
 58# isort: on
 59# ##-- end types
 60
 61##-- logging
 62logging = logmod.getLogger(__name__)
 63##-- end logging
 64
 65RUN_DRY_SWITCH                               = doot.constants.decorations.RUN_DRY_SWITCH
 66GEN_TASKS                                    = doot.constants.decorations.GEN_TASKS
 67IO_ACT                                       = doot.constants.decorations.IO_ACT
 68CONTROL_FLOW                                 = doot.constants.decorations.CONTROL_FLOW
 69EXTERNAL                                     = doot.constants.decorations.EXTERNAL
 70STATE_MOD                                    = doot.constants.decorations.STATE_MOD
 71ANNOUNCER                                    = doot.constants.decorations.ANNOUNCER
 72
 73dry_run_active                               = doot.args.on_fail(False).cmd.args.dry_run()
 74##--|
 75

[docs]
 76class _BaseMetaAction(IdempotentDec):
 77

[docs]
 78    def _wrap_fn_h(self, fn):
 79        return self._wrap_method_h(fn)

 80

[docs]
 81    def _wrap_class_h(self, cls):
 82        cls.__call__ = self._wrap_method_h(cls.__call__)
 83        return cls


 84

[docs]
 85class DryRunSwitch(_BaseMetaAction):
 86    """ Mark an action callable/class as to be skipped in dry runs """
 87
 88    def __init__(self, *, override:bool=False):
 89        super().__init__("dry_run", mark="_dry_run_mark")
 90        self._override = override or dry_run_active
 91

[docs]
 92    def _wrap_method_h(self, fn):
 93        override_active = self._override or dry_run_active
 94
 95        def _can_disable(*args, **kwargs):
 96            if override_active:
 97                return None
 98            return fn(*args, **kwargs)
 99
100        return _can_disable


101

[docs]
102class GeneratesTasks(_BaseMetaAction):
103    """ Mark an action callable/class as a task generator """
104
105    def __init__(self):
106        super().__init__(GEN_TASKS, mark="_gen_data_mark")
107

[docs]
108    def _wrap_fn_h(self, fn):
109
110        def _gen_task_wrapper(*args, **kwargs):
111            match fn(*args, **kwargs):
112                case [*xs] if any(not isinstance(x, SpecStruct_p) for x in xs):
113                    raise doot.errors.ActionCallError("Action did not return task specs")
114                case list() as res:
115                    return res
116                case _:
117                    raise doot.errors.ActionCallError("Action did not return a list of generated tasks")
118
119        return _gen_task_wrapper


120

[docs]
121class IOWriter(_BaseMetaAction):
122    """ mark an action callable/class as an io action,
123      checks the path it'll write to isn't write protected
124    """
125
126    def __init__(self, *, targets=None):
127        super().__init__(IO_ACT, mark="_io_mark")
128        self._targets = targets or ["to"]
129

[docs]
130    def _wrap_fn_h(self, fn):
131
132        def _io_writer_wrapper(*args, **kwargs):
133            result = fn(*args, **kwargs)
134            raise NotImplementedError()
135
136        return _io_writer_wrapper


137

[docs]
138class ControlFlow(_BaseMetaAction):
139    """ mark an action callable/class as a control flow action
140      implies it runs dry
141      """
142
143    def __init__(self):
144        super().__init__(CONTROL_FLOW, mark="_control_flow_mark")

145

[docs]
146class External(_BaseMetaAction):
147    """ mark an action callable/class as calling an external program.
148      implies rundryswitch
149      """
150
151    def __init__(self):
152        super().__init__(EXTERNAL, mark="_external_mark")

153

[docs]
154class StateManipulator(_BaseMetaAction):
155    """ mark an action callable/class as a state modifier
156      checks the DootKey `returns` are in the return dict
157    """
158
159    def __init__(self):
160        super().__init__(STATE_MOD, mark="_state_mod_mark")

161

[docs]
162class Announcer(_BaseMetaAction):
163    """ mark an action callable/class as reporting in a particular way
164      implies run_dry, and skips on cli arg `silent`
165      """
166
167    def __init__(self):
168        super().__init__(ANNOUNCER, mark="_announcer_mark")