1#!/usr/bin/env python3
2"""
3
4"""
5# ruff: noqa: ERA001
6# Import:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import datetime
11import sys
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# ##-- end stdlib imports
23
24from jgdv.debugging import TraceBuilder
25from ._core import IdempotentDec, MetaDec, MonotonicDec
26
27# ##-- types
28# isort: off
29import abc
30import collections.abc
31from typing import TYPE_CHECKING, cast, assert_type, assert_never
32from typing import Generic, NewType, Concatenate
33# Protocols:
34from typing import Protocol, runtime_checkable
35# Typing Decorators:
36from typing import no_type_check, final, override, overload
37
38if TYPE_CHECKING:
39 from . import _interface as API # noqa: N812
40
41 from jgdv import Maybe, Either, Method, Func
42 from typing import Final
43 from typing import ClassVar, Any, LiteralString
44 from typing import Never, Self, Literal
45 from typing import TypeGuard, ParamSpec
46 from collections.abc import Iterable, Iterator, Callable, Generator
47 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
48 type Logger = logmod.Logger
49
50# isort: on
51# ##-- end types
52
53##-- logging
54logging = logmod.getLogger(__name__)
55##-- end logging
56
57# Vars:
58
59# Body:
60
[docs]
61class NoSideEffects(MetaDec):
62 """ TODO Mark a Target as not modifying external variables """
63 pass
64
[docs]
65class CanRaise(MetaDec):
66 """ TODO mark a target as able to raise certain errors.
67 Non-exaustive, doesn't change runtime behaviour,
68 just to simplify documentation
69
70 """
71 pass
72
[docs]
73class Breakpoint(IdempotentDec):
74 """
75 Decorator to attach a breakpoint to a function, without pausing execution
76 """
77
78 @override
79 def __call__[**I, O](self, target:Callable[I,O]) -> Callable[I,O]:
80 msg = "needs RunningDebugger"
81 raise NotImplementedError(msg)
82
83 # # TODO handle repeats
84 # if args[0].breakpoint:
85
86 # f_code = f.__code__
87 # db = RunningDebugger()
88 # # Ensure trace function is set
89 # sys.settrace(db.trace_dispatch)
90 # if not db.get_break(f_code.co_filename, f_code.co_firstlineno+2):
91 # db.set_break(f_code.co_filename,
92 # f_code.co_firstlineno+2,
93 # True)
94 # else:
95 # bp = Breakpoint.bplist[f_code.co_filename,
96 # f_code.co_firstlineno+2][0]
97 # bp.enable()
98
99 # return self._func(self, *args, **kwargs)
100
[docs]
101class MethodMaybe(MonotonicDec):
102 """ Make a fn or method propagate None's """
103 type MethMb[Obj,X,**I,O] = Callable[Concatenate[Obj, X, I],O]
104 type FuncMb[X,**I,O] = Callable[Concatenate[X, I],O]
105 # def __call__[**I, X,Obj, O](self, target:MethMb[Obj,X,I,O]|FuncMb[X,I,O], *args:Any, **kwargs:Any) -> MethMb[Obj,Maybe[X],I,O]|FuncMb[Maybe[X],I,O]: # type: ignore[override]
106
107 @override
108 def __call__[Obj,**I,X,O](self, target:MethMb[Obj,X,I,O], *args:Any, **kwargs:Any) -> MethMb[Obj,Maybe[X],I,O]: # type: ignore[override]
109 return MonotonicDec.__call__(self, target, *args, **kwargs)
110
[docs]
111 @override
112 def _wrap_method_h[Obj, X, **I, O](self, meth:MethMb[Obj,X,I,Maybe[O]]) -> MethMb[Obj,Maybe[X],I,Maybe[O]]: # type: ignore[override]
113
114 def _prop_maybe(_self:Obj, fst:Maybe[X], *args:I.args, **kwargs:I.kwargs) -> Maybe[O]:
115 match fst:
116 case None:
117 return None
118 case x:
119 return meth(_self, x, *args, **kwargs)
120
121 return _prop_maybe
122
123
[docs]
124class FnMaybe(MonotonicDec):
125 """ Make a fn or method propagate None's """
126 type MethMb[Obj,X,**I,O] = Callable[Concatenate[Obj, X, I],O]
127 type FuncMb[X,**I,O] = Callable[Concatenate[X, I],O]
128 # def __call__[**I, X,Obj, O](self, target:MethMb[Obj,X,I,O]|FuncMb[X,I,O], *args:Any, **kwargs:Any) -> MethMb[Obj,Maybe[X],I,O]|FuncMb[Maybe[X],I,O]: # type: ignore[override]
129
130 @override
131 def __call__[**I,X,O](self, target:FuncMb[X,I,O], *args:Any, **kwargs:Any) -> FuncMb[Maybe[X],I,O]: # type: ignore[override]
132 return MonotonicDec.__call__(self, target, *args, **kwargs)
133
[docs]
134 @override
135 def _wrap_fn_h[X, **I, O](self, fn:FuncMb[X, I, Maybe[O]]) -> FuncMb[Maybe[X], I, Maybe[O]]: # type: ignore[override]
136
137 def _prop_maybe(fst:Maybe[X], *args:I.args, **kwargs:I.kwargs) -> Maybe[O]:
138 match fst:
139 case None:
140 return None
141 case x:
142 try:
143 return fn(x, *args, **kwargs)
144 except Exception as err:
145 err.with_traceback(TraceBuilder[2:]) # type: ignore[misc]
146 raise
147
148 return _prop_maybe
149
150
[docs]
151class DoEither(MonotonicDec):
152 """ Either do the fn/method, or propagate the error """
153
[docs]
154 @override
155 def _wrap_method_h[X,Y, **I, O, E:Exception](self, meth:Callable[Concatenate[X, Y, I], Either[O,E]]) -> API.Decorated[Concatenate[X, Y|E, I], Either[O, E]]: # type: ignore[override]
156
157 def _prop_either(_self:X, fst:Y|E, *args:I.args, **kwargs:I.kwargs) -> Either[O, E]:
158 match fst:
159 case Exception() as err:
160 return cast("E", err)
161 case x:
162 try:
163 return meth(_self, x, *args, **kwargs)
164 except Exception as err: # noqa: BLE001
165 err.with_traceback(TraceBuilder[2:]) # type: ignore[misc]
166 return cast("E", err)
167
168 return _prop_either
169
[docs]
170 @override
171 def _wrap_fn_h[X, **I, O, E:Exception](self, fn:Func[Concatenate[X, I], O]) -> Func[Concatenate[X|E, I], Either[O, E]]: # type: ignore[override]
172
173 def _prop_either(fst:X|E, *args:I.args, **kwargs:I.kwargs) -> Either[O, E]:
174 match fst:
175 case Exception() as err:
176 return cast("E", err)
177 case x:
178 try:
179 return fn(x, *args, **kwargs)
180 except Exception as err: # noqa: BLE001
181 err.with_traceback(TraceBuilder[2:]) # type: ignore[misc]
182 return cast("E", err)
183
184 return _prop_either