前言

本長文不適合手機端閱讀，請酌情退出

公司架構組基于 pytest 自研了一套測試框架 sstest，目的是為了讓業務組（也就是我在的組）更好的寫單元測試，從而提高代碼質量，單元測試的目的是為了回歸校驗，避免新提交的代碼影響到項目中舊的功能。

我是組里第一個接入 sstest 的同學，踩了很多坑... 從而對 pytest 的源碼產生了興趣，在閱讀 pytest 源碼的過程中，發現 pluggy 插件系統其實是 pytest 的核心，可以說 pytest 只是將多個插件利用 pluggy 構建出來的項目，所以先分析 pluggy。

老規矩，在開始分析前，希望自己搞清楚的幾個問題：1. 如何使用 pluggy？

2. 插件代碼如何做到靈活可插拔的？

3. 外部系統如何調用插件邏輯？

隨著分析的進行會有新的問題拋出，問題可以幫助我們理清目的，避免迷失在源碼中。

整體把控

pluggy 插件系統與我此前研究的 python 插件系統不同，pluggy 不可以動態插入，即無法在程序運行的過程中利用插件添加新的功能。

pluggy 主要有 3 個概念：1.PluginManager：用于管理插件規范與插件本身

2.HookspecMarker：定義插件調用規范，每個規范可以對應 1~N 個插件，每個插件都滿足該規范，否則無法成功被外部調用

3.HookimplMarker：定義插件，插件邏輯具體的實現在該類裝飾的方法中

簡單使用一下，代碼如下。

import pluggy

# 創建插件規范類裝飾器

hookspac = pluggy.HookspecMarker('example')

# 創建插件類裝飾器

hookimpl = pluggy.HookimplMarker('example')

class MySpec(object):

# 創建插件規范

@hookspac

def myhook(self, a, b):

pass

class Plugin_1(object):

# 定義插件

@hookimpl

def myhook(self, a, b):

return a + b

class Plugin_2(object):

@hookimpl

def myhook(self, a, b):

return a - b

# 創建manger和添加hook規范

pm = pluggy.PluginManager('example')

pm.add_hookspecs(MySpec)

# 注冊插件

pm.register(Plugin_1())

pm.register(Plugin_2())

# 調用插件中的myhook方法

results = pm.hook.myhook(a=10, b=20)

print(results)

整段代碼簡單而言就是創建相應的類裝飾器裝飾類中的方法，通過這些類裝飾器構建出了插件規范與插件本身。

首先，實例化 PluginManager 類，實例化時需要傳入全局唯一的 project name，HookspecMarker 類與 HookimplMarker 類的實例化也需要使用相同的 project name。

創建完插件管理器后，通過 add_hookspecs 方法添加插件規范、通過 register 方法添加插件本身則可。

添加完插件調用規范與插件本身后，就可以通過插件管理器的 hook 屬性直接調用插件了。

閱讀到這里，關于問題「1，2，3」便有了答案。

pluggy 使用的過程可以分為 4 步：1. 通過 HookspecMarker 類裝飾器定義插件調用規范

2. 通過 HookimplMarker 類裝飾器定義插件邏輯

3. 創建 PluginManager 并綁定插件調用規范與插件本身

4. 調用插件

通過類裝飾器與 PluginManager.add_hookspecs、PluginManager.register 方法的配合，輕松實現插件的可插拔操作，其背后原理其實就是被類裝飾器裝飾后的方法會被動態添加上新的屬性信息，而對應的 add_hookspecs 與 register 等方法會根據這些屬性信息來判斷是否為插件規范或插件本身。

想要在外部系統中使用插件，只需要調用 pm.hook.any_hook_function 方法則可，任意注冊了插件都可以被輕松調用。

但這里引出了新的問題：4. 類裝飾器是如何將某個類中的方法設置成插件的？

5.pluggy 是如何關聯插件規范與插件本身的？

6. 插件中的邏輯具體是如何被調用的？

這三個問題關注的是實現細節，下面進一步步進行分析。

hookspac 與 hookimpl 裝飾器的作用

代碼中，使用了 hookspac 類裝飾器定義出插件調用規范，使用了 hookimpl 類裝飾器定義出插件本身，兩者的作用其實就是「為被裝飾的方法添加新的屬性」。因為兩者邏輯相似，所以這里就只分析 hookspac 類裝飾器代碼，代碼如下：

class HookspecMarker(object):

def __init__(self, project_name):

self.project_name = project_name

def __call__(

self, function=None, firstresult=False, historic=False, warn_on_impl=None

):

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

類裝飾器主要會重寫類的__call__方法，上述代碼中__call__方法最核心的邏輯便是使用 setattr 方法為被裝飾的 func 方法添加新的屬性，屬性名為當前 project name 加上_spec 后綴，而屬性的值為一個字典對象。

在調用 PluginManager.add_hookspecs 方法時會利用 hookspac 類裝飾器添加的信息

HookimplMarker 類類似，只是添加的屬性有所不同，核心代碼如下。

setattr(

func,

self.project_name + "_impl",

dict(

hookwrapper=hookwrapper,

optionalhook=optionalhook,

tryfirst=tryfirst,

trylast=trylast,

specname=specname,

),

)

所以關于「4. 類裝飾器是如何將某個類中的方法設置成插件的？」，其實就是利用 setattr 方法為當前方法設置新的屬性，這些屬性相當于提供了一種信息，PluginManager 會根據這些信息判斷該方法是不是插件，跟下面例子本質相同。

In [1]: def fuc1():

...: print('hh')

...:

In [2]: setattr(fuc1, 'fuc1' + '_impl', dict(a=1, b=2))

In [3]: fuc1.fuc1_impl

Out[3]: {'a': 1, 'b': 2}

添加插件規范與注冊插件的背后

實例化 pluggy.PluginManager 類后便可以通過 add_hookspecs 方法添加插件規范與 register 方法注冊插件。

要搞清楚「pluggy 是如何關聯插件規范與插件本身的？」，就需要深入它們的源碼。

實例化 PluginManager 類，其實就是調用它的__init__方法。

def __init__(self, project_name, implprefix=None):

"""If ``implprefix`` is given implementation functions

will be recognized if their name matches the ``implprefix``. """

self.project_name = project_name

# ...省略部分...

# 關鍵

self.hook = _HookRelay()

self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(

methods,

kwargs,

firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,

)

關鍵在于定義了 self.hook 屬性與 self._inner_hookexec 屬性，它是一個匿名方法，會接收 hook、methods、kwargs 三個參數并將這些參數傳遞給 hook.multicall 方法。

隨后調用 add_hookspecs 方法添加插件規范，其代碼如下。

class PluginManager(object):

# 獲取被裝飾方法中對應屬性的信息（HookspecMarker裝飾器添加的信息）

def parse_hookspec_opts(self, module_or_class, name):

method = getattr(module_or_class, name)

return getattr(method, self.project_name + "_spec", None)

def add_hookspecs(self, module_or_class):

names = []

for name in dir(module_or_class):

# 獲取插件規范信息

spec_opts = self.parse_hookspec_opts(module_or_class, name)

if spec_opts is not None:

hc = getattr(self.hook, name, None)

if hc is None:

hc = _HookCaller(name, self._hookexec, module_or_class, spec_opts)

setattr(self.hook, name, hc)

# ...省略部分代碼...

上述代碼中通過 parse_hookspec_opts 方法獲取方法中相應屬性的參數，如果參數不為 None 那么則獲取_HookRelay 類中的被裝飾方法的信息（該方法就是 MySpec 類的 myhook），從源碼中可以發現_HookRelay 類其實是空類，它存在的意義其實就是接收新的屬性，分析到后面你就會發現_HookRelay 類其實就是用于連接插件規范與插件本身的類。

如果_HookRelay 類中沒有 myhook 屬性信息，則實例化_HookCaller 類并將其作為 self.hook 的屬性，具體而言，就是將_HookCaller 類的實例作為_HookRelay 類中 myhook 屬性的值。

_HookCaller 類很重要，其部分代碼如下。

class _HookCaller(object):

def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):

self.name = name

# ...省略...

self._hookexec = hook_execute

self.spec = None

if specmodule_or_class is not None:

assert spec_opts is not None

self.set_specification(specmodule_or_class, spec_opts)

def has_spec(self):

return self.spec is not None

def set_specification(self, specmodule_or_class, spec_opts):

assert not self.has_spec()

self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)

if spec_opts.get("historic"):

self._call_history = []

關鍵在于 set_specification 方法，該方法會實例化 HookSpec 類并將其復制給 self.spec。

至此，插件規范就添加完了，緊接著通過 register 方法注冊插件本身，其核心代碼如下。

def register(self, plugin, name=None):

# 省略部分代碼

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) # 獲取 hookspec 插件規范

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)

首先通過 self.parse_hookimpl_opts 方法獲取被 hookimpl 裝飾器添加的信息，隨后通過 getattr (plugin, name) 方法獲取方法名，其實就是 myhook，最后初始化 HookImpl 類，其實就是插件本身，并將其與對應的插件規范綁定，通過_add_hookimpl 方法實現這一目的。

_add_hookimpl 方法會根據 hookimpl 實例中的屬性判斷其插入的位置，不同位置，調用順序不同，代碼如下。

def _add_hookimpl(self, hookimpl):

"""Add an implementation to the callback chain.

"""

# 是否有 包裝器 (即插件邏輯中使用了yield關鍵字)

if hookimpl.hookwrapper:

methods = self._wrappers

else:

methods = self._nonwrappers

# 先調用還是后代碼

if hookimpl.trylast:

methods.insert(0, hookimpl)

elif hookimpl.tryfirst:

methods.append(hookimpl)

else:

# find last non-tryfirst method

i = len(methods) - 1

while i >= 0 and methods[i].tryfirst:

i -= 1

methods.insert(i + 1, hookimpl)

至此「5.pluggy 是如何關聯插件規范與插件本身的？」的問題也是明白了，簡單而言，插件規范與插件本身都被裝飾器添加了特殊信息，通過這些特殊信息將兩者找到并分布利用這些屬性的值初始化_HookCaller 類（插件規范）與 HookImpl 類（插件本身），最后通過_add_hookimpl 方法完成綁定。

插件中的邏輯具體是如何被調用的？

從一開始的示例代碼中，可以發現，調用 myhook 插件方法通過 pm.hook.myhook(a=10, b=20) 方法則可。

背后是什么？

PluginManager.hook 其實就是_HookRelay 類，而_HookRelay 類模式是一個空類，通過 add_hookspecs 方法與 register 方法的操作，_HookRelay 類中多了名為 myhook 的屬性，該屬性對應著_HookCaller 類實例。

pm.hook.myhook (a=10, b=20) 其實就是調用_HookCaller.__call__，代碼如下。

def __call__(self, *args, **kwargs):

# 省略部分代碼

if self.spec and self.spec.argnames:

# 計算插件規范中可以接收的參數與插件本身可以接收的參數是否相同

notincall = (

set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())

)

if notincall:

# 省略代碼

# 調用方法

return self._hookexec(self, self.get_hookimpls(), kwargs)

__call__方法的主要就是判斷插件規范與插件本身是否匹配，然后通過 self._hookexec 方法去真正的執行。

通過分析，完整的調用鏈條為：_HookCaller._hookexec -> PluginManager._inner_hookexec -> _HookCaller.multicall -> callers文件的中的_multicall方法

_multicall 方法中最關鍵的代碼片段如下。

def _multicall(hook_impls, caller_kwargs, firstresult=False):

for hook_impl in reversed(hook_impls):

try:

# 調用myhook方法

args = [caller_kwargs[argname] for argname in hook_impl.argnames]

# 省略代碼

# 如果插件中使用了yeild，則通過這種方式調用

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")

至此，pluggy 的核心邏輯就擼完了。

尾部

如果你看完了，恭喜呀，但這只是 pluggy 最簡單的模式，它還有一些比較重要的方法，因為篇幅原因就沒往上貼了，各位感興趣可以自行研究或跟我探討。

后面抽空出篇水本談談看源碼的一些建議。

總結

以上是生活随笔為你收集整理的基于python3写的源码剖析_Python 极简插件系统 pluggy 源码剖析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。