1#!/usr/bin/env python3
2"""
3
4"""
5# mypy: disable-error-code="attr-defined"
6# ruff: noqa: ANN002, ANN003
7# Imports:
8from __future__ import annotations
9
10# ##-- stdlib imports
11import datetime
12import enum
13import functools as ftz
14import itertools as itz
15import keyword
16import logging as logmod
17import pathlib as pl
18import re
19import time
20import types as types_
21import typing
22import weakref
23from uuid import UUID, uuid1
24
25# ##-- end stdlib imports
26
27# ##-- 1st party imports
28from jgdv import Mixin
29from jgdv.decorators import (
30 DataDec,
31 MetaDec,
32 DecoratorAccessor_m,
33 DForm_e,
34)
35from jgdv.structs.strang import CodeReference
36from .. import errors as dkey_errs
37from ..dkey import DKey
38from .._interface import ARGS_K, KWARGS_K, PARAM_IGNORES, Key_p
39
40# ##-- end 1st party imports
41
42# ##-- types
43# isort: off
44import abc
45import collections.abc
46from typing import TYPE_CHECKING, Generic, cast, assert_type, assert_never
47# Protocols:
48from typing import Protocol, runtime_checkable
49# Typing Decorators:
50from typing import no_type_check, final, override, overload
51from types import MethodType
52from collections.abc import Mapping
53from typing import Any
54
55if TYPE_CHECKING:
56 from jgdv import Method
57 import inspect
58 from jgdv import Decorator, FmtStr, Func, Ident, Maybe, Rx
59 from typing import Final
60 from typing import ClassVar, LiteralString
61 from typing import Never, Self, Literal
62 from typing import TypeGuard
63 from collections.abc import Iterable, Iterator, Callable, Generator
64 from collections.abc import Sequence, MutableMapping, Hashable
65
66 from jgdv.decorators._interface import Decorated, Decorable
67 type Signature = inspect.Signature
68# isort: on
69# ##-- end types
70
71##-- logging
72logging = logmod.getLogger(__name__)
73##-- end logging
74
83
[docs]
84class DKeyExpansionDecorator(DataDec):
85 """
86 Utility class for idempotently decorating actions with auto-expanded keys
87
88 """
89 _param_ignores : tuple[str, ...]
90
91 def __init__(self, keys:list[DKey], ignores:Maybe[list[str]]=None, **kwargs) -> None:
92 kwargs.setdefault("mark", "_dkey_marked")
93 kwargs.setdefault("data", "_dkey_vals")
94 super().__init__(keys, **kwargs) # type: ignore[arg-type]
95 match ignores:
96 case None:
97 self._param_ignores = PARAM_IGNORES
98 case list():
99 self._param_ignores = tuple(ignores)
100 case x:
101 raise TypeError(type(x))
102
[docs]
103 @override
104 def _wrap_method_h[**In, Out](self, meth:Func[In,Out]) -> Decorated[In, Out]:
105 data_key = self._data_key
106
107 def _method_action_expansions(*call_args:In.args, **kwargs:In.kwargs) -> Out:
108 _self, spec, state, *rest = call_args
109 try:
110 expansions = [x(spec, state) for x in meth.__annotations__[data_key]]
111 except KeyError as err:
112 logging.warning("Action State Expansion Failure: %s", err)
113 return cast("Out", False) # noqa: FBT003
114 else:
115 return meth(*call_args, *expansions, **kwargs)
116
117 # -
118 return cast("Decorated[In, Out]", _method_action_expansions)
119
[docs]
120 @override
121 def _wrap_fn_h[**In, Out](self, fn:Func[In, Out]) -> Decorated[In, Out]:
122 data_key = self._data_key
123
124 def _fn_action_expansions(*args:In.args, **kwargs:In.kwargs) -> Out:
125 spec, state, *rest = args
126 try:
127 expansions = [x(spec, state) for x in fn.__annotations__[data_key]]
128 except KeyError as err:
129 logging.warning("Action State Expansion Failure: %s", err)
130 return cast("Out", False) # noqa: FBT003
131 else:
132 return fn(*args, *expansions, **kwargs)
133
134 # -
135 return _fn_action_expansions
136
[docs]
137 @override
138 def _validate_sig_h(self, sig:Signature, form:DForm_e, args:Maybe[list[DKey]]=None) -> None:
139 x : Any
140 y : Any
141 ##--|
142 # Get the head args
143 match form:
144 case DForm_e.FUNC:
145 head = ["spec", "state"]
146 case DForm_e.METHOD:
147 head = ["self", "spec", "state"]
148 case x:
149 raise TypeError(type(x))
150
151 params = list(sig.parameters)
152 tail = args or []
153
154 # Check the head
155 for x,y in zip(params, head, strict=False):
156 if x != y:
157 msg = "Mismatch in signature head"
158 raise dkey_errs.DecorationMismatch(msg, x, y, form)
159
160 prefix_ig, suffix_ig = self._param_ignores
161 # Then the tail, backwards, because the decorators are applied in reverse order
162 for x,y in zip(params[::-1], tail[::-1], strict=False):
163 assert(isinstance(y, Key_p))
164 key_str = y.var_name()
165 if x.startswith(prefix_ig) or x.endswith(suffix_ig):
166 logging.debug("Skipping: %s", x)
167 continue
168
169 if keyword.iskeyword(key_str):
170 msg = "Key is a keyword, use an alias like _{} or {}_ex, or use named={}"
171 raise dkey_errs.DecorationMismatch(msg, x, key_str)
172
173 if not key_str.isidentifier():
174 msg = "Key is not an identifier, use an alias _{} or {}_ex or use named={}"
175 raise dkey_errs.DecorationMismatch(msg, x, key_str)
176
177 if x != y:
178 msg = "Mismatch in signature tail"
179 raise dkey_errs.DecorationMismatch(msg, str(x), key_str)
180
[docs]
181class DKeyed:
182 """ Decorators for actions
183
184 It registers arguments on an action and extracts them from the spec and state automatically.
185
186 provides: expands/paths/types/requires/returns/args/kwargs/redirects
187 The kwarg 'hint' takes a dict and passes the contents to the relevant expansion method as kwargs
188
189 arguments are added to the tail of the action args, in order of the decorators.
190 the name of the expansion is expected to be the name of the action parameter,
191 with a "_" prepended if the name would conflict with a keyword., or with "_ex" as a suffix
192 eg: @DKeyed.paths("from") -> def __call__(self, spec, state, _from):...
193 or: @DKeyed.paths("from") -> def __call__(self, spec, state, from_ex):...
194 """
195
196 _extensions : ClassVar[set[type]] = set()
197 _decoration_builder : ClassVar[type] = DKeyExpansionDecorator
198
199 @override
200 def __init_subclass__(cls) -> None:
201 """
202 Subclasses of DKeyed are stored, and used to extend DKeyed
203 """
204 super().__init_subclass__()
205 if cls in DKeyed._extensions:
206 return
207 DKeyed._extensions.add(cls)
208 for x in dir(cls):
209 match getattr(cls, x):
210 case MethodType() as y if not hasattr(DKeyed, x):
211 setattr(DKeyed, x, y)
212 case _:
213 pass
214
231
[docs]
232class DKeyedRetrieval(DecoratorAccessor_m, DKeyed):
233 """ Subclass extension for DKeyed decorators,
234 which modify the calling behaviour of the decoration target
235
236 """
237
238 _decoration_builder : ClassVar[type] = DKeyExpansionDecorator
239
244
[docs]
245 @classmethod
246 def expands(cls, *args, **kwargs) -> Decorator:
247 """ mark an action as using expanded string keys """
248 return cls.formats(*args, **kwargs)
249
[docs]
250 @classmethod
251 def paths(cls, *args, **kwargs) -> Decorator:
252 """ mark an action as using expanded path keys """
253 kwargs.setdefault("implicit", True)
254 keys = [DKey[pl.Path](x, **kwargs) for x in args]
255 return cls._build_decorator(keys)
256
[docs]
257 @classmethod
258 def types(cls, *args, **kwargs) -> Decorator:
259 """ mark an action as using raw type keys """
260 keys : list = [DKey[Any](x, implicit=True, **kwargs) for x in args]
261 return cls._build_decorator(keys)
262
[docs]
263 @classmethod
264 def toggles(cls, *args, **kwargs) -> Decorator:
265 keys : list = [DKey(x, implicit=True, ctor=bool, check=bool, **kwargs) for x in args]
266 return cls._build_decorator(keys)
267
[docs]
268 @classmethod
269 def args(cls, fn:Callable) -> Decorator:
270 """ mark an action as using spec.args """
271 keys = [DKey[DKey.Marks.ARGS](ARGS_K, implicit=True)] # type: ignore[name-defined]
272 match cls._build_decorator(keys)(fn):
273 case None:
274 raise dkey_errs.DecorationMismatch()
275 case x:
276 return x
277
[docs]
278 @classmethod
279 def kwargs(cls, fn:Callable) -> Decorator:
280 """ mark an action as using all kwargs"""
281 keys = [DKey[DKey.Marks.KWARGS](KWARGS_K, implicit=True)] # type: ignore[name-defined]
282 match cls._build_decorator(keys)(fn):
283 case None:
284 raise dkey_errs.DecorationMismatch()
285 case x:
286 return x
287
[docs]
288 @classmethod
289 def redirects(cls, *args, **kwargs) -> Decorator:
290 """ mark an action as using redirection keys """
291 kwargs.setdefault("max_exp", 1)
292 keys = [DKey[Mapping](x, implicit=True, **kwargs) for x in args] # type: ignore[name-defined]
293 return cls._build_decorator(keys)
294
[docs]
295 @classmethod
296 def references(cls, *args, **kwargs) -> Decorator:
297 """ mark keys to use as to_coderef imports """
298 keys = [DKey[CodeReference](x, implicit=True, **kwargs) for x in args] # type: ignore[name-defined]
299 return cls._build_decorator(keys)
300
[docs]
301 @classmethod
302 def postbox(cls, *args, **kwargs) -> Decorator:
303 keys = [DKey[DKey.Marks.POSTBOX](x, implicit=True, **kwargs) for x in args] # type: ignore[name-defined]
304 return cls._build_decorator(keys)