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
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
|
"""Call loop machinery."""
import sys
from types import TracebackType
from typing import Any, List, Optional, Tuple, Type, Union
from .exceptions import HookCallError, PluginCallError
from .implementation import HookImplementation
def _raise_wrapfail(wrap_controller, msg):
co = wrap_controller.gi_code
raise RuntimeError(
"wrap_controller at %r %s:%d %s"
% (co.co_name, co.co_filename, co.co_firstlineno, msg)
)
ExcInfo = Union[
Tuple[Type[BaseException], BaseException, TracebackType],
Tuple[None, None, None],
]
class HookResult:
"""A class to store/modify results from a :func:`~hooks._multicall` hook loop.
Results are accessed in ``.result`` property, which will also raise
any exceptions that occured during the hook loop.
Parameters
----------
results : List[Tuple[Any, HookImplementation]]
A list of (result, HookImplementation) tuples, with the result and
HookImplementation object responsible for each result collected during
a _multicall loop.
excinfo : tuple
The output of sys.exc_info() if raised during the multicall loop.
firstresult : bool, optional
Whether the hookspec had ``firstresult == True``, by default False.
If True, self._result, and self.implementation will be single values,
otherwise they will be lists.
plugin_errors : list
A list of any :class:`PluginCallError` instances that were created
during the multicall loop.
"""
def __init__(
self,
result: List[Tuple[Any, HookImplementation]],
excinfo: Optional[ExcInfo],
firstresult: bool = False,
plugin_errors: Optional[List[PluginCallError]] = None,
):
self._result: Any = []
#: The HookImplementation(s) that were responsible for each result in ``result``
self.implementation: Optional[
Union[HookImplementation, List[HookImplementation]]
] = []
#: Whether this HookResult came from a ``firstresult`` multicall.
self.is_firstresult: bool = firstresult
self._excinfo = excinfo
self.plugin_errors = plugin_errors
if result:
self._result, self.implementation = tuple(zip(*result))
self._result = list(self._result)
if firstresult:
if self._result:
self._result = self._result[0]
self.implementation = self.implementation[0] # type: ignore
else:
self._result = None
self.implementation = None
#: Name of last hookwrapper that changed the result, if any
self._modified_by: Optional[str] = None
@property
def excinfo(self):
return self._excinfo
@classmethod
def from_call(cls, func):
"""Used when hookcall monitoring is enabled.
https://pluggy.readthedocs.io/en/latest/#call-monitoring
"""
__tracebackhide__ = True
try:
return func()
except BaseException:
return cls(None, sys.exc_info())
def force_result(self, result: Any):
"""Force the result(s) to ``result``.
This may be used by hookwrappers to alter this result object.
If the hook was marked as a ``firstresult`` a single value should
be set otherwise set a (modified) list of results. Any exceptions
found during invocation will be deleted.
"""
import inspect
self._result = result
self._excinfo = None
self._modified_by = inspect.stack()[1].function
@property
def result(self) -> Union[Any, List[Any]]:
"""Return the result(s) for this hook call.
If the hook was marked as a ``firstresult`` only a single value
will be returned otherwise a list of results.
"""
__tracebackhide__ = True
if self._excinfo is not None:
_type, value, traceback = self._excinfo
if value:
raise value.with_traceback(traceback)
return self._result
def _multicall(
hook_impls: List[HookImplementation],
caller_kwargs: dict,
firstresult: bool = False,
) -> HookResult:
"""The primary :class:`~napari_plugin_engine.HookImplementation` call loop.
Parameters
----------
hook_impls : list
A sequence of hook implementation (HookImplementation) objects
caller_kwargs : dict
Keyword:value pairs to pass to each ``hook_impl.function``. Every
key in the dict must be present in the ``argnames`` property for each
``hook_impl`` in ``hook_impls``.
firstresult : bool, optional
If ``True``, return the first non-null result found, otherwise, return
a list of results from all hook implementations, by default False
Returns
-------
outcome : HookResult
A :class:`HookResult` object that contains the results returned by
plugins along with other metadata about the call.
Raises
------
HookCallError
If one or more of the keys in ``caller_kwargs`` is not present in one
of the ``hook_impl.argnames``.
PluginCallError
If ``firstresult == True`` and a plugin raises an Exception.
"""
__tracebackhide__ = True
results = []
errors: List['PluginCallError'] = []
excinfo: Optional[ExcInfo] = None
try: # run impl and wrapper setup functions in a loop
teardowns = []
try:
for hook_impl in reversed(hook_impls):
# skip disabled hook implementations
if not getattr(hook_impl, 'enabled', True):
continue
args: List[Any] = []
try:
args = [
caller_kwargs[argname]
for argname in hook_impl.argnames
]
except KeyError:
raise HookCallError(
"hook call must provide argument the following "
f"arguments: {set(hook_impl.argnames)!r}, but provided"
f" {set(caller_kwargs)!r}"
)
if hook_impl.hookwrapper:
try:
gen = hook_impl(*args)
next(gen) # first yield
teardowns.append(gen)
except StopIteration:
_raise_wrapfail(gen, "did not yield")
else:
res = None
# this is where the plugin function actually gets called
# we put it in a try/except so that if one plugin throws
# an exception, we don't lose the whole loop
try:
res = hook_impl(*args)
except Exception as exc:
# creating a PluginCallError will store it for later
# in plugins.exceptions.PLUGIN_ERRORS
errors.append(PluginCallError(hook_impl, cause=exc))
# if it was a `firstresult` hook, break and raise now.
if firstresult:
break
if res is not None:
results.append((res, hook_impl))
if firstresult: # halt further impl calls
break
except BaseException:
excinfo = sys.exc_info()
finally:
if firstresult and errors:
raise errors[-1]
outcome = HookResult(
results,
excinfo=excinfo,
firstresult=firstresult,
plugin_errors=errors,
)
# run all wrapper post-yield blocks
for gen in reversed(teardowns):
try:
gen.send(outcome)
_raise_wrapfail(gen, "has second yield")
except StopIteration:
pass
return outcome
|
