Pytest 源码解读 [2] - [pluggy] 核心设计理念和代码结构

代码结构

pluggy 的核心代码非常简介,把 repo 克隆到本地目录后,它的核心代码就是在 src目录下的4个文件

  • _tracing.py - 调试作用,把 hookspec 和 plugin 调用链路的分析打印出来,一般不开启
  • callers.py - 主要是 _multicall 这个方法,用来实现对于 plugin 的调用逻辑
  • hooks.py - 核心是 HookspeckMarkerHookimplMarker, 用来装饰 hook 和 plugin 的方法
  • manager.py - PluginManager 是整个 pluggy 的核心类,负责 hook 和 plugin 的管理和核心调用逻辑

核心设计理念

整个 pluggy 的核心逻辑就是这么几行代码

1
2
3
4
5
pm = PluginManager("pluggy_demo_1")
pm.add_hookspecs(HookSpec)
pm.register(HookImpl1())
pm.register(HookImpl2())
print(pm.hook.calculate(a=1, b=2))
  • 创建一个 PluginManager 对象
  • 调用 add_hookspecs,注册一个新的 hook object
  • 调用 register 注册一个新的 plugin object
  • 通过 pm.hook.calculate 来实现对 plugin 的调用

这里就有几个核心设计的问题,通过解答这些问题可以帮助我们对于整个 pluggy的核心设计原理有进一步的了解

  1. pluggy 是怎么知道 Spec 里面哪些方法是定义的 hook ?

  2. pluggy 是怎么把 Plugin 的具体实现和 hook 做关联的

  3. 实际调用的时候,pm.hook 是个什么, 如何实现最终的 plugin 调用的

pluggy 是怎么知道 Spec 里面哪些方法是定义的 hook ?

这个就要全靠 HookspecMarkerHookimplMarker 这两个装饰器了,因为两者的实现逻辑类似,这里就只解释一下HookspecMarker, 我们先看上篇博文的demo中,我们一开始就定义了2个装饰器,并用他们来注释了 hook的方法

1
2
3
4
5
6
7
8
spec = HookspecMarker("pluggy_demo_1")
impl = HookimplMarker("pluggy_demo_1")


class HookSpec:
@spec(firstresult=True)
def calculate(self, a, b):
pass

这里HookspecMarker是一个基于class实现的装饰器,我们看一下代码里它的具体实现逻辑 , __call__方法,

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
def __call__(
self, function=None, firstresult=False, historic=False, warn_on_impl=None
):
""" if passed a function, directly sets attributes on the function
which will make it discoverable to add_hookspecs(). If passed no
function, returns a decorator which can be applied to a function
later using the attributes supplied.

If firstresult is True the 1:N hook call (N being the number of registered
hook implementation functions) will stop at I<=N when the I'th function
returns a non-None result.

If historic is True calls to a hook will be memorized and replayed
on later registered plugins.

"""

def setattr_hookspec_opts(func):
if historic and firstresult:
raise ValueError("cannot have a historic firstresult hook")
setattr(
func,
self.project_name + "_spec",
dict(
firstresult=firstresult,
historic=historic,
warn_on_impl=warn_on_impl,
),
)
return func

if function is not None:
return setattr_hookspec_opts(function)
else:
return setattr_hookspec_opts

原来主要的功能就是给被装饰的函数增加一个特别的属性, 属性的名字是 Project_name + _spec, 属性的value 就是装饰器的参数取值,

这样在PluginMangeradd_hookspecs函数的逻辑中,我们就可以看到通过遍历对象的所以属性,找到含有这这个特殊的attribute 的方法就好了

1
2
3
for name in dir(module_or_class):
spec_opts = self.parse_hookspec_opts(module_or_class, name)
if spec_opts is not None:

当然这里就有一个前提,全局的project_name 必须是一致的且唯一,因为这个key是拼出来的,如果不一致就找不到 对应的 hook 和 impl 定义了

pluggy 是怎么把 Plugin 的具体实现和 hook 做关联的 ?

首先我们先看一下 pm.hook, 这个 hook 是个什么东西,我们在 PluginManager 的构造函数看到了hook的定义

1
self.hook = _HookRelay()

看了一下_HookReplay 的代码,只有一个类的定义,看起来就是一个 ,这里可以猜测一下,当我们真实调用pm.hook.calculate(a=1,b=2) 的时候,实际的调用可能是这样的

1
getattr(pm.hook, "calculate")(a=1, b=2)

为了印证我的猜测,我在 PluginManagerregister方法这找到了这段逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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)
hook = getattr(self.hook, name, None)
if hook is None:
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)
  • 遍历 plugin 对象的所有属性(method), 通过 self.parse_hookimpl_opts这个方法找到含有特殊 attribute 的方法,就是被 impl 装饰器装饰过的方法
  • 然后再把一个_HookCaller的对象添加到hook对象中, setattr(self.hook, name, hook), 所以看起来真实的调用是 _HookCaller(a=1, b=2), 具体我们下次再分析
  • 如何实现1:N的调用关系呢,实际上一个hook 可以添加多个 hookimpl, 看这行代码就清楚了,它是在找到了 hook 绑定后调用的 hook._add_hookimpl(hookimpl)

实际调用的时候,pm.hook 是个什么, 如何实现最终的 plugin 调用的

前面说了实际上 pm.hook.calculate获取的是一个 _HookCaller对象,所以真实的是调用了它的__call__方法,我们简单看一下它的逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
if self.spec and self.spec.argnames:
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(tuple(notincall)),
stacklevel=2,
)
return self._hookexec(self, self.get_hookimpls(), kwargs)

核心代码就是最后一行, self._hookexec ,把具体的kwargs传入,来调用实际的 plugin 实现,具体的这个 self._hookexec方法,是在构造_HookCaller的时候作为一个参数传入的,它的定义看起来也只是一个封装

1
2
3
4
def _hookexec(self, hook, methods, kwargs):
# called from all hookcaller instances.
# enable_tracing will set its own wrapping function at self._inner_hookexec
return self._inner_hookexec(hook, methods, kwargs)

真正的逻辑是 PluginManger构造时候封装的 _multicall函数,这个具体我们下次再分析它是如何实现的

1
2
3
4
5
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
)

总结

到这里我们把pluggy主要的设计思路都介绍完了,后面会对于几个核心类的实现做进一步的分析,欢迎有兴趣的同学和我留言