Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/pluggy/hooks.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:4f3585e2f14b
1 """
2 Internal hook annotation, representation and calling machinery.
3 """
4 import inspect
5 import sys
6 import warnings
7 from .callers import _legacymulticall, _multicall
8
9
10 class HookspecMarker(object):
11 """ Decorator helper class for marking functions as hook specifications.
12
13 You can instantiate it with a project_name to get a decorator.
14 Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions
15 if the :py:class:`.PluginManager` uses the same project_name.
16 """
17
18 def __init__(self, project_name):
19 self.project_name = project_name
20
21 def __call__(
22 self, function=None, firstresult=False, historic=False, warn_on_impl=None
23 ):
24 """ if passed a function, directly sets attributes on the function
25 which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`.
26 If passed no function, returns a decorator which can be applied to a function
27 later using the attributes supplied.
28
29 If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered
30 hook implementation functions) will stop at I<=N when the I'th function
31 returns a non-``None`` result.
32
33 If ``historic`` is ``True`` calls to a hook will be memorized and replayed
34 on later registered plugins.
35
36 """
37
38 def setattr_hookspec_opts(func):
39 if historic and firstresult:
40 raise ValueError("cannot have a historic firstresult hook")
41 setattr(
42 func,
43 self.project_name + "_spec",
44 dict(
45 firstresult=firstresult,
46 historic=historic,
47 warn_on_impl=warn_on_impl,
48 ),
49 )
50 return func
51
52 if function is not None:
53 return setattr_hookspec_opts(function)
54 else:
55 return setattr_hookspec_opts
56
57
58 class HookimplMarker(object):
59 """ Decorator helper class for marking functions as hook implementations.
60
61 You can instantiate with a ``project_name`` to get a decorator.
62 Calling :py:meth:`.PluginManager.register` later will discover all marked functions
63 if the :py:class:`.PluginManager` uses the same project_name.
64 """
65
66 def __init__(self, project_name):
67 self.project_name = project_name
68
69 def __call__(
70 self,
71 function=None,
72 hookwrapper=False,
73 optionalhook=False,
74 tryfirst=False,
75 trylast=False,
76 ):
77
78 """ if passed a function, directly sets attributes on the function
79 which will make it discoverable to :py:meth:`.PluginManager.register`.
80 If passed no function, returns a decorator which can be applied to a
81 function later using the attributes supplied.
82
83 If ``optionalhook`` is ``True`` a missing matching hook specification will not result
84 in an error (by default it is an error if no matching spec is found).
85
86 If ``tryfirst`` is ``True`` this hook implementation will run as early as possible
87 in the chain of N hook implementations for a specification.
88
89 If ``trylast`` is ``True`` this hook implementation will run as late as possible
90 in the chain of N hook implementations.
91
92 If ``hookwrapper`` is ``True`` the hook implementations needs to execute exactly
93 one ``yield``. The code before the ``yield`` is run early before any non-hookwrapper
94 function is run. The code after the ``yield`` is run after all non-hookwrapper
95 function have run. The ``yield`` receives a :py:class:`.callers._Result` object
96 representing the exception or result outcome of the inner calls (including other
97 hookwrapper calls).
98
99 """
100
101 def setattr_hookimpl_opts(func):
102 setattr(
103 func,
104 self.project_name + "_impl",
105 dict(
106 hookwrapper=hookwrapper,
107 optionalhook=optionalhook,
108 tryfirst=tryfirst,
109 trylast=trylast,
110 ),
111 )
112 return func
113
114 if function is None:
115 return setattr_hookimpl_opts
116 else:
117 return setattr_hookimpl_opts(function)
118
119
120 def normalize_hookimpl_opts(opts):
121 opts.setdefault("tryfirst", False)
122 opts.setdefault("trylast", False)
123 opts.setdefault("hookwrapper", False)
124 opts.setdefault("optionalhook", False)
125
126
127 if hasattr(inspect, "getfullargspec"):
128
129 def _getargspec(func):
130 return inspect.getfullargspec(func)
131
132
133 else:
134
135 def _getargspec(func):
136 return inspect.getargspec(func)
137
138
139 _PYPY3 = hasattr(sys, "pypy_version_info") and sys.version_info.major == 3
140
141
142 def varnames(func):
143 """Return tuple of positional and keywrord argument names for a function,
144 method, class or callable.
145
146 In case of a class, its ``__init__`` method is considered.
147 For methods the ``self`` parameter is not included.
148 """
149 cache = getattr(func, "__dict__", {})
150 try:
151 return cache["_varnames"]
152 except KeyError:
153 pass
154
155 if inspect.isclass(func):
156 try:
157 func = func.__init__
158 except AttributeError:
159 return (), ()
160 elif not inspect.isroutine(func): # callable object?
161 try:
162 func = getattr(func, "__call__", func)
163 except Exception:
164 return (), ()
165
166 try: # func MUST be a function or method here or we won't parse any args
167 spec = _getargspec(func)
168 except TypeError:
169 return (), ()
170
171 args, defaults = tuple(spec.args), spec.defaults
172 if defaults:
173 index = -len(defaults)
174 args, kwargs = args[:index], tuple(args[index:])
175 else:
176 kwargs = ()
177
178 # strip any implicit instance arg
179 # pypy3 uses "obj" instead of "self" for default dunder methods
180 implicit_names = ("self",) if not _PYPY3 else ("self", "obj")
181 if args:
182 if inspect.ismethod(func) or (
183 "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names
184 ):
185 args = args[1:]
186
187 try:
188 cache["_varnames"] = args, kwargs
189 except TypeError:
190 pass
191 return args, kwargs
192
193
194 class _HookRelay(object):
195 """ hook holder object for performing 1:N hook calls where N is the number
196 of registered plugins.
197
198 """
199
200
201 class _HookCaller(object):
202 def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
203 self.name = name
204 self._wrappers = []
205 self._nonwrappers = []
206 self._hookexec = hook_execute
207 self.argnames = None
208 self.kwargnames = None
209 self.multicall = _multicall
210 self.spec = None
211 if specmodule_or_class is not None:
212 assert spec_opts is not None
213 self.set_specification(specmodule_or_class, spec_opts)
214
215 def has_spec(self):
216 return self.spec is not None
217
218 def set_specification(self, specmodule_or_class, spec_opts):
219 assert not self.has_spec()
220 self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
221 if spec_opts.get("historic"):
222 self._call_history = []
223
224 def is_historic(self):
225 return hasattr(self, "_call_history")
226
227 def _remove_plugin(self, plugin):
228 def remove(wrappers):
229 for i, method in enumerate(wrappers):
230 if method.plugin == plugin:
231 del wrappers[i]
232 return True
233
234 if remove(self._wrappers) is None:
235 if remove(self._nonwrappers) is None:
236 raise ValueError("plugin %r not found" % (plugin,))
237
238 def get_hookimpls(self):
239 # Order is important for _hookexec
240 return self._nonwrappers + self._wrappers
241
242 def _add_hookimpl(self, hookimpl):
243 """Add an implementation to the callback chain.
244 """
245 if hookimpl.hookwrapper:
246 methods = self._wrappers
247 else:
248 methods = self._nonwrappers
249
250 if hookimpl.trylast:
251 methods.insert(0, hookimpl)
252 elif hookimpl.tryfirst:
253 methods.append(hookimpl)
254 else:
255 # find last non-tryfirst method
256 i = len(methods) - 1
257 while i >= 0 and methods[i].tryfirst:
258 i -= 1
259 methods.insert(i + 1, hookimpl)
260
261 if "__multicall__" in hookimpl.argnames:
262 warnings.warn(
263 "Support for __multicall__ is now deprecated and will be"
264 "removed in an upcoming release.",
265 DeprecationWarning,
266 )
267 self.multicall = _legacymulticall
268
269 def __repr__(self):
270 return "<_HookCaller %r>" % (self.name,)
271
272 def __call__(self, *args, **kwargs):
273 if args:
274 raise TypeError("hook calling supports only keyword arguments")
275 assert not self.is_historic()
276 if self.spec and self.spec.argnames:
277 notincall = (
278 set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
279 )
280 if notincall:
281 warnings.warn(
282 "Argument(s) {} which are declared in the hookspec "
283 "can not be found in this hook call".format(tuple(notincall)),
284 stacklevel=2,
285 )
286 return self._hookexec(self, self.get_hookimpls(), kwargs)
287
288 def call_historic(self, result_callback=None, kwargs=None, proc=None):
289 """Call the hook with given ``kwargs`` for all registered plugins and
290 for all plugins which will be registered afterwards.
291
292 If ``result_callback`` is not ``None`` it will be called for for each
293 non-``None`` result obtained from a hook implementation.
294
295 .. note::
296 The ``proc`` argument is now deprecated.
297 """
298 if proc is not None:
299 warnings.warn(
300 "Support for `proc` argument is now deprecated and will be"
301 "removed in an upcoming release.",
302 DeprecationWarning,
303 )
304 result_callback = proc
305
306 self._call_history.append((kwargs or {}, result_callback))
307 # historizing hooks don't return results
308 res = self._hookexec(self, self.get_hookimpls(), kwargs)
309 if result_callback is None:
310 return
311 # XXX: remember firstresult isn't compat with historic
312 for x in res or []:
313 result_callback(x)
314
315 def call_extra(self, methods, kwargs):
316 """ Call the hook with some additional temporarily participating
317 methods using the specified ``kwargs`` as call parameters. """
318 old = list(self._nonwrappers), list(self._wrappers)
319 for method in methods:
320 opts = dict(hookwrapper=False, trylast=False, tryfirst=False)
321 hookimpl = HookImpl(None, "<temp>", method, opts)
322 self._add_hookimpl(hookimpl)
323 try:
324 return self(**kwargs)
325 finally:
326 self._nonwrappers, self._wrappers = old
327
328 def _maybe_apply_history(self, method):
329 """Apply call history to a new hookimpl if it is marked as historic.
330 """
331 if self.is_historic():
332 for kwargs, result_callback in self._call_history:
333 res = self._hookexec(self, [method], kwargs)
334 if res and result_callback is not None:
335 result_callback(res[0])
336
337
338 class HookImpl(object):
339 def __init__(self, plugin, plugin_name, function, hook_impl_opts):
340 self.function = function
341 self.argnames, self.kwargnames = varnames(self.function)
342 self.plugin = plugin
343 self.opts = hook_impl_opts
344 self.plugin_name = plugin_name
345 self.__dict__.update(hook_impl_opts)
346
347 def __repr__(self):
348 return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin)
349
350
351 class HookSpec(object):
352 def __init__(self, namespace, name, opts):
353 self.namespace = namespace
354 self.function = function = getattr(namespace, name)
355 self.name = name
356 self.argnames, self.kwargnames = varnames(function)
357 self.opts = opts
358 self.argnames = ["__multicall__"] + list(self.argnames)
359 self.warn_on_impl = opts.get("warn_on_impl")