hook調(diào)用過程源碼解析
調(diào)用過程主要涉及到：
- callers 模塊中的 _multicall 函數(shù)
- hook 模塊中的 _HookCaller 類的 __call__ 方法
當我們 add_hookspecs hookspec 以及 register hookimpl
通過 PluginManager.hook.myhook(**kwargs) 之后：
-
調(diào)用myhook(實際就是一個 _HookCaller 類的一個實例)的 __call__ 方法
__call__ 作用：檢查是否是 historic (historic=True時，是需要通過myhook.call_historic方式傳入?yún)?shù)，在register時，自動調(diào)用)，實際傳遞參數(shù)與 hookspec 是否一致 (不一致并不會導致運行停止，而是 raise 一個UserWarning)，實際傳遞參數(shù)是否都是鍵值對。
最終調(diào)用 _hookexec 傳入相應的參數(shù)。
class _HookCaller:
# 構(gòu)造函數(shù)，實例化傳入 hook_execute
def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
self.name = name
self._wrappers = []
self._nonwrappers = []
self._hookexec = hook_execute
self._call_history = None
self.spec = None
if specmodule_or_class is not None:
assert spec_opts is not None
self.set_specification(specmodule_or_class, spec_opts)
# 調(diào)用初始化時傳入的 hook_execute 函數(shù)
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
# This is written to avoid expensive operations when not needed.
# 判斷實際調(diào)用的hook傳入的參數(shù)是否與hookspec一致，否則rasie UserWarning
if self.spec:
for argname in self.spec.argnames:
if argname not in kwargs:
notincall = tuple(set(self.spec.argnames) - kwargs.keys())
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(notincall),
stacklevel=2,
)
break
firstresult = self.spec.opts.get("firstresult")
else:
firstresult = False
# __call__ 的作用，檢查是否是historic，是否實際參數(shù)與hookspec一致，是否傳入?yún)?shù)都是鍵值對，
# 實際上是調(diào)用了callers中的_multicall函數(shù)，
# self.get_hookimpls() 返回所有hooimpl，hookwrapper 在后，nonwrapper在前
# 等價于 _multicall(self.name, self.get_hookimpls(), kwargs, firstresult)
return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
-
__call__ 之后，會調(diào)用實例化 _HookCaller 時傳入的 hook_execute (本質(zhì)上是一個 callers 模塊中的 _multicall 函數(shù))
- _HookCaller 對象是在 register 時實例化的，并且通過 setattr 設置 _HookRelay 對象的一個屬性，通過該方式存儲所有的 hookimpl ，即 1:N 個
class PluginManager:
def __init__(self, project_name):
self.project_name = project_name
self._name2plugin = {}
self._plugin2hookcallers = {}
self._plugin_distinfo = []
self.trace = _tracing.TagTracer().get("pluginmanage")
self.hook = _HookRelay()
self._inner_hookexec = _multicall
def _hookexec(self, hook_name, methods, kwargs, firstresult):
# called from all hookcaller instances.
# enable_tracing will set its own wrapping function at self._inner_hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
def register(self, plugin, name=None):
""" Register a plugin and return its canonical name or ``None`` if the name
is blocked from registering. Raise a :py:class:`ValueError` if the plugin
is already registered. """
plugin_name = name or self.get_canonical_name(plugin)
if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
if self._name2plugin.get(plugin_name, -1) is None:
return # blocked plugin, return None to indicate no registration
raise ValueError(
"Plugin already registered: %s=%s\n%s"
% (plugin_name, plugin, self._name2plugin)
)
# XXX if an error happens we should make sure no state has been
# changed at point of return
self._name2plugin[plugin_name] = plugin
# register matching hook implementations of the plugin
self._plugin2hookcallers[plugin] = hookcallers = []
for name in dir(plugin):
hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
if hookimpl_opts is not None:
normalize_hookimpl_opts(hookimpl_opts)
method = getattr(plugin, name)
hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
name = hookimpl_opts.get("specname") or name
hook = getattr(self.hook, name, None)
if hook is None:
# hook = _HookCaller(name, self._hookexec)
# setattr(self.hook, name, hook)
# _HookCaller 對象是在 register 時實例化的，并且通過 setattr 設置 _HookRelay 對象的一個屬性
hook = _HookCaller(name, self._hookexec)
setattr(self.hook, name, hook)
elif hook.has_spec():
self._verify_hook(hook, hookimpl)
hook._maybe_apply_history(hookimpl)
hook._add_hookimpl(hookimpl)
hookcallers.append(hook)
return plugin_name
-
_multicall 的作用：
通過 reversed 函數(shù)反轉(zhuǎn) hook_impls 列表 (get_hookimpls 方法返回的列表，hookwrapper 在后，nonwrapper 在前)，所以hookwrapper 優(yōu)先調(diào)用，遍歷所有 register 的 hookimpl 對象，并調(diào)用相應的 function
如果 hookimpl 對象是一個 hookwrapper (即 hookwrapper=True)，則調(diào)用相應的 function，返回一個生成器，并 next() 迭代一次, 將此時的迭代狀態(tài)通過一個 teardowns list 維護, 這里并不會獲取 hookwrapper 的返回值，也并不會受 firstresult=True 影響
如果 hookimpl 對象是一個 nonwrapper, 調(diào)用 function，并將返回值增加到一個 result 列表中，如果 firstresult=True，則 break 遍歷
遍歷調(diào)用結(jié)束后，將實例一個 _Result 對象，傳入 result，并且反轉(zhuǎn) teardowns，繼續(xù)執(zhí)行 yield 之后的代碼，之前先調(diào)用的 hookwrapper，現(xiàn)在最后調(diào)用。最終返回 _Result.get_result()，即 result 列表
def _multicall(hook_name, hook_impls, caller_kwargs, firstresult):
"""Execute a call into multiple python functions/methods and return the
result(s).
``caller_kwargs`` comes from _HookCaller.__call__().
"""
__tracebackhide__ = True
results = []
excinfo = None
try: # run impl and wrapper setup functions in a loop
teardowns = []
try:
# 通過 reversed 函數(shù)反轉(zhuǎn) hook_impls 列表
# get_hookimpls 方法返回的列表，hookwrapper 在后，nonwrapper 在前
# 所以hookwrapper 優(yōu)先調(diào)用
for hook_impl in reversed(hook_impls):
try:
args = [caller_kwargs[argname] for argname in hook_impl.argnames]
except KeyError:
for argname in hook_impl.argnames:
if argname not in caller_kwargs:
raise HookCallError(
"hook call must provide argument %r" % (argname,)
)
# 如果 hookimpl 對象是一個 hookwrapper (即 hookwrapper=True)，
# 則調(diào)用相應的 function，返回一個生成器，并 next() 迭代一次,
# 將此時的迭代狀態(tài)通過一個 teardowns list 維護,
# 這里并不會獲取 hookwrapper 的返回值，也并不會受 firstresult=True 影響
# 如果 hookimpl 對象是一個 nonwrapper, 調(diào)用 function，
# 并將返回值增加到一個 result 列表中，如果 firstresult=True，則 break 遍歷
if hook_impl.hookwrapper:
try:
gen = hook_impl.function(*args)
next(gen) # first yield
teardowns.append(gen)
except StopIteration:
_raise_wrapfail(gen, "did not yield")
else:
res = hook_impl.function(*args)
if res is not None:
results.append(res)
if firstresult: # halt further impl calls
break
except BaseException:
excinfo = sys.exc_info()
# 遍歷調(diào)用結(jié)束后，將實例一個 _Result 對象，傳入 result，并且反轉(zhuǎn) teardowns，繼續(xù)執(zhí)行 yield 之后的代碼
# 之前先調(diào)用的 hookwrapper，現(xiàn)在最后調(diào)用。最終返回 _Result.get_result()，即 result 列表
finally:
if firstresult: # first result hooks return a single value
outcome = _Result(results[0] if results else None, excinfo)
else:
outcome = _Result(results, excinfo)
# 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.get_result()
