Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pluggy/_hooks.py: 61%
156 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1"""
2Internal hook annotation, representation and calling machinery.
3"""
4import inspect
5import sys
6import warnings
9class HookspecMarker:
10 """Decorator helper class for marking functions as hook specifications.
12 You can instantiate it with a project_name to get a decorator.
13 Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions
14 if the :py:class:`.PluginManager` uses the same project_name.
15 """
17 def __init__(self, project_name):
18 self.project_name = project_name
20 def __call__(
21 self, function=None, firstresult=False, historic=False, warn_on_impl=None
22 ):
23 """if passed a function, directly sets attributes on the function
24 which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`.
25 If passed no function, returns a decorator which can be applied to a function
26 later using the attributes supplied.
28 If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered
29 hook implementation functions) will stop at I<=N when the I'th function
30 returns a non-``None`` result.
32 If ``historic`` is ``True`` calls to a hook will be memorized and replayed
33 on later registered plugins.
35 """
37 def setattr_hookspec_opts(func):
38 if historic and firstresult:
39 raise ValueError("cannot have a historic firstresult hook")
40 setattr(
41 func,
42 self.project_name + "_spec",
43 dict(
44 firstresult=firstresult,
45 historic=historic,
46 warn_on_impl=warn_on_impl,
47 ),
48 )
49 return func
51 if function is not None:
52 return setattr_hookspec_opts(function)
53 else:
54 return setattr_hookspec_opts
57class HookimplMarker:
58 """Decorator helper class for marking functions as hook implementations.
60 You can instantiate with a ``project_name`` to get a decorator.
61 Calling :py:meth:`.PluginManager.register` later will discover all marked functions
62 if the :py:class:`.PluginManager` uses the same project_name.
63 """
65 def __init__(self, project_name):
66 self.project_name = project_name
68 def __call__(
69 self,
70 function=None,
71 hookwrapper=False,
72 optionalhook=False,
73 tryfirst=False,
74 trylast=False,
75 specname=None,
76 ):
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.
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).
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.
89 If ``trylast`` is ``True`` this hook implementation will run as late as possible
90 in the chain of N hook implementations.
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).
99 If ``specname`` is provided, it will be used instead of the function name when
100 matching this hook implementation to a hook specification during registration.
102 """
104 def setattr_hookimpl_opts(func):
105 setattr(
106 func,
107 self.project_name + "_impl",
108 dict(
109 hookwrapper=hookwrapper,
110 optionalhook=optionalhook,
111 tryfirst=tryfirst,
112 trylast=trylast,
113 specname=specname,
114 ),
115 )
116 return func
118 if function is None:
119 return setattr_hookimpl_opts
120 else:
121 return setattr_hookimpl_opts(function)
124def normalize_hookimpl_opts(opts):
125 opts.setdefault("tryfirst", False)
126 opts.setdefault("trylast", False)
127 opts.setdefault("hookwrapper", False)
128 opts.setdefault("optionalhook", False)
129 opts.setdefault("specname", None)
132_PYPY = hasattr(sys, "pypy_version_info")
135def varnames(func):
136 """Return tuple of positional and keywrord argument names for a function,
137 method, class or callable.
139 In case of a class, its ``__init__`` method is considered.
140 For methods the ``self`` parameter is not included.
141 """
142 if inspect.isclass(func):
143 try:
144 func = func.__init__
145 except AttributeError:
146 return (), ()
147 elif not inspect.isroutine(func): # callable object?
148 try:
149 func = getattr(func, "__call__", func)
150 except Exception:
151 return (), ()
153 try: # func MUST be a function or method here or we won't parse any args
154 spec = inspect.getfullargspec(func)
155 except TypeError:
156 return (), ()
158 args, defaults = tuple(spec.args), spec.defaults
159 if defaults:
160 index = -len(defaults)
161 args, kwargs = args[:index], tuple(args[index:])
162 else:
163 kwargs = ()
165 # strip any implicit instance arg
166 # pypy3 uses "obj" instead of "self" for default dunder methods
167 implicit_names = ("self",) if not _PYPY else ("self", "obj")
168 if args:
169 if inspect.ismethod(func) or (
170 "." in getattr(func, "__qualname__", ()) and args[0] in implicit_names
171 ):
172 args = args[1:]
174 return args, kwargs
177class _HookRelay:
178 """hook holder object for performing 1:N hook calls where N is the number
179 of registered plugins.
181 """
184class _HookCaller:
185 def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
186 self.name = name
187 self._wrappers = []
188 self._nonwrappers = []
189 self._hookexec = hook_execute
190 self._call_history = None
191 self.spec = None
192 if specmodule_or_class is not None:
193 assert spec_opts is not None
194 self.set_specification(specmodule_or_class, spec_opts)
196 def has_spec(self):
197 return self.spec is not None
199 def set_specification(self, specmodule_or_class, spec_opts):
200 assert not self.has_spec()
201 self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
202 if spec_opts.get("historic"):
203 self._call_history = []
205 def is_historic(self):
206 return self._call_history is not None
208 def _remove_plugin(self, plugin):
209 def remove(wrappers):
210 for i, method in enumerate(wrappers):
211 if method.plugin == plugin:
212 del wrappers[i]
213 return True
215 if remove(self._wrappers) is None:
216 if remove(self._nonwrappers) is None:
217 raise ValueError(f"plugin {plugin!r} not found")
219 def get_hookimpls(self):
220 # Order is important for _hookexec
221 return self._nonwrappers + self._wrappers
223 def _add_hookimpl(self, hookimpl):
224 """Add an implementation to the callback chain."""
225 if hookimpl.hookwrapper:
226 methods = self._wrappers
227 else:
228 methods = self._nonwrappers
230 if hookimpl.trylast:
231 methods.insert(0, hookimpl)
232 elif hookimpl.tryfirst:
233 methods.append(hookimpl)
234 else:
235 # find last non-tryfirst method
236 i = len(methods) - 1
237 while i >= 0 and methods[i].tryfirst:
238 i -= 1
239 methods.insert(i + 1, hookimpl)
241 def __repr__(self):
242 return f"<_HookCaller {self.name!r}>"
244 def __call__(self, *args, **kwargs):
245 if args:
246 raise TypeError("hook calling supports only keyword arguments")
247 assert not self.is_historic()
249 # This is written to avoid expensive operations when not needed.
250 if self.spec:
251 for argname in self.spec.argnames:
252 if argname not in kwargs:
253 notincall = tuple(set(self.spec.argnames) - kwargs.keys())
254 warnings.warn(
255 "Argument(s) {} which are declared in the hookspec "
256 "can not be found in this hook call".format(notincall),
257 stacklevel=2,
258 )
259 break
261 firstresult = self.spec.opts.get("firstresult")
262 else:
263 firstresult = False
265 return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
267 def call_historic(self, result_callback=None, kwargs=None):
268 """Call the hook with given ``kwargs`` for all registered plugins and
269 for all plugins which will be registered afterwards.
271 If ``result_callback`` is not ``None`` it will be called for for each
272 non-``None`` result obtained from a hook implementation.
273 """
274 self._call_history.append((kwargs or {}, result_callback))
275 # Historizing hooks don't return results.
276 # Remember firstresult isn't compatible with historic.
277 res = self._hookexec(self.name, self.get_hookimpls(), kwargs, False)
278 if result_callback is None:
279 return
280 for x in res or []:
281 result_callback(x)
283 def call_extra(self, methods, kwargs):
284 """Call the hook with some additional temporarily participating
285 methods using the specified ``kwargs`` as call parameters."""
286 old = list(self._nonwrappers), list(self._wrappers)
287 for method in methods:
288 opts = dict(hookwrapper=False, trylast=False, tryfirst=False)
289 hookimpl = HookImpl(None, "<temp>", method, opts)
290 self._add_hookimpl(hookimpl)
291 try:
292 return self(**kwargs)
293 finally:
294 self._nonwrappers, self._wrappers = old
296 def _maybe_apply_history(self, method):
297 """Apply call history to a new hookimpl if it is marked as historic."""
298 if self.is_historic():
299 for kwargs, result_callback in self._call_history:
300 res = self._hookexec(self.name, [method], kwargs, False)
301 if res and result_callback is not None:
302 result_callback(res[0])
305class HookImpl:
306 def __init__(self, plugin, plugin_name, function, hook_impl_opts):
307 self.function = function
308 self.argnames, self.kwargnames = varnames(self.function)
309 self.plugin = plugin
310 self.opts = hook_impl_opts
311 self.plugin_name = plugin_name
312 self.__dict__.update(hook_impl_opts)
314 def __repr__(self):
315 return f"<HookImpl plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>"
318class HookSpec:
319 def __init__(self, namespace, name, opts):
320 self.namespace = namespace
321 self.function = function = getattr(namespace, name)
322 self.name = name
323 self.argnames, self.kwargnames = varnames(function)
324 self.opts = opts
325 self.warn_on_impl = opts.get("warn_on_impl")