/
mixins.py
206 lines (158 loc) · 5.96 KB
/
mixins.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
from _py_abc import ABCMeta
from abc import abstractmethod
from typing import List
from pypads.utils.util import is_package_available
DEFAULT_ORDER = 1
class NoCallAllowedError(Exception):
"""
Exception to denote that a callable couldn't be called, but isn't essential.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
class MissingDependencyError(NoCallAllowedError):
"""
Exception to be thrown if a dependency is missing.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
class SuperStop:
"""
This class resolves the issue TypeError: object.__init__() takes exactly one argument by being the last class
in a mro and ommitting all arguments. This should be always last in the mro()!
"""
def __init__(self, *args, **kwargs):
mro = self.__class__.mro()
if SuperStop in mro:
if len(mro) - 2 != mro.index(SuperStop):
raise ValueError("SuperStop ommitting arguments in " + str(self.__class__)
+ " super() callstack: " + str(mro))
super().__init__()
class OrderMixin(SuperStop):
"""
Object defining an order attribute to denote its priority.
"""
__metaclass__ = ABCMeta
@abstractmethod
def __init__(self, *args, order=DEFAULT_ORDER, **kwargs):
self._order = order
super().__init__(*args, **kwargs)
def order(self):
return self._order
@staticmethod
def sort(collection, reverse=False): # type: (List[OrderMixin]) -> List[OrderMixin]
copy = collection.copy()
copy.sort(key=lambda e: e.order(), reverse=reverse)
return copy
@staticmethod
def sort_mutable(collection, reverse=False): # type: (List[OrderMixin]) -> None
collection.sort(key=lambda e: e.order(), reverse=reverse)
class CallableMixin(SuperStop):
"""
Object defining a _call method which can be overwritten.
"""
__metaclass__ = ABCMeta
@abstractmethod
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
return self.__real_call__(*args, **kwargs)
@abstractmethod
def __real_call__(self, *args, **kwargs):
pass
class DependencyMixin(CallableMixin):
"""
Callable being able to be disabled / enabled depending on package availability.
"""
__metaclass__ = ABCMeta
@abstractmethod
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@staticmethod
@abstractmethod
def _needed_packages():
"""
Overwrite this to provide your package names.
:return: List of needed packages by the logger.
"""
return []
def _check_dependencies(self):
"""
Raise error if dependencies are missing.
"""
missing = []
packages = self._needed_packages()
if packages is not None:
for package in packages:
if not is_package_available(package):
missing.append(package)
if len(missing) > 0:
raise MissingDependencyError("Can't log " + str(self) + ". Missing dependencies: " + ", ".join(missing))
def __call__(self, *args, **kwargs):
self._check_dependencies()
return super().__call__(*args, **kwargs)
class IntermediateCallableMixin(CallableMixin):
"""
Callable being able to be disable / enabled on nested / intermediate runs.
"""
__metaclass__ = ABCMeta
@abstractmethod
def __init__(self, *args, nested=True, intermediate=True, **kwargs):
self._intermediate = intermediate
self._nested = nested
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
from pypads.app.pypads import is_nested_run
if self._nested or not is_nested_run():
from pypads.app.pypads import is_intermediate_run
if self._intermediate or not is_intermediate_run():
return super().__call__(*args, **kwargs)
raise NoCallAllowedError("Call wasn't allowed by intermediate / nested settings of the current run.")
@property
def nested(self):
return self._nested
@property
def intermediate(self):
return self._intermediate
class TimedCallableMixin(CallableMixin):
__metaclass__ = ABCMeta
"""
Callable tracking its own execution time.
"""
@abstractmethod
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
c = super().__call__
from pypads.injections.analysis.time_keeper import timed
_return, time = timed(lambda: c(*args, **kwargs))
return _return, time
class DefensiveCallableMixin(CallableMixin):
__metaclass__ = ABCMeta
"""
Callable handling errors produced by itself.
"""
@abstractmethod
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __call__(self, ctx, *args, _pypads_env=None, **kwargs):
try:
return super().__call__(ctx, *args, _pypads_env=_pypads_env, **kwargs)
except KeyboardInterrupt:
return self._handle_error(*args, ctx=ctx, _pypads_env=_pypads_env, error=Exception("KeyboardInterrupt"),
**kwargs)
except Exception as e:
return self._handle_error(*args, ctx=ctx, _pypads_env=_pypads_env, error=e, **kwargs)
@abstractmethod
def _handle_error(self, *args, ctx, _pypads_env, error, **kwargs):
raise NotImplementedError()
class ConfigurableCallableMixin(CallableMixin):
__metaclass__ = ABCMeta
"""
Callable storing additional creation args as fields to be accessible later on.
"""
@abstractmethod
def __init__(self, *args, **kwargs):
self._kwargs = kwargs
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
super().__call__(*args, **{**self._kwargs, **kwargs})