用pluggy实现动态载入模块

pluggy是pytest用的一个插件系统, 通过pluggy我们可以实现动载入插件模块功能.

用 pluggy.HookspecMarker 声明一个spec, 确定一个插件需要实现哪些方法.

这里我们定义了一个str2str的实现规范, 它需要一个process的函数用于处理字符串.


import pluggy
import pathlib
import importlib

hookspec = pluggy.HookspecMarker("str2str")

class Str2StrSpec:
    @hookspec
    def process(self, input_str: str) -> str:
        """Process input data and return output data."""
        pass

新建一个PluginManager, 并指定plugin manager需要的hookspec

注意这个plugin manager的名子也需要和实现规范保持一致, 都要叫"str2str"


pm = pluggy.PluginManager("str2str")
pm.add_hookspecs(Str2StrSpec)

新建一些插件.

在plugins目录下, 我们可以新建一些python文件实现相关的插件. 这里我们用@hookimpl显示指定了实现规范str2str所需的函数.

为了能够方便的自动加载, 我们写了一个register的方法, 调用这个方法可以把插件加入到plugin manager中. 注意这里需要手动指定插件的name, 否则register的时候会自动生成一个随机数做为name.


import pluggy

hookimpl = pluggy.HookimplMarker("str2str")


class UpperCasePlugin:
    @hookimpl
    def process(self, input_str: str) -> str:
        return input_str.upper()
        
plugin_name = 'upper'

def register(pm):
    pm.register(UpperCasePlugin(), name='upper')

import pluggy

hookimpl = pluggy.HookimplMarker("str2str")


class ReversePlugin:
    @hookimpl
    def process(self, input_str: str) -> str:
        return input_str[::-1]

plugin_name = 'reverse'

def register(pm):
    pm.register(ReversePlugin(), name='reverse')

通过遍历目录下的py文件, 调用其中的register方法就可以把插件加载的plugin manager中. 随后, 我们可以用list_name_plugin显示插件的名子, 并可以手动调用特定插件.


# 遍历 `plugins` 目录，自动导入并注册
plugin_dir = pathlib.Path("plugins")
for plugin_file in plugin_dir.glob("*.py"):
    if plugin_file.stem != "__init__":
        module_name = f"plugins.{plugin_file.stem}"
        if module_name in sys.modules:
            module = importlib.reload(sys.modules[module_name])
        else:
            module = importlib.import_module(module_name)
        if hasattr(module, "register"):
            plugin_name = module.plugin_name
            pm.unregister(name=plugin_name)
            module.register(pm)
            
print("已发现插件:", pm.list_name_plugin())

pm.get_plugin(name='upper').process('hello')
pm.get_plugin(name='reverse').process('hello')

稍加扩展, 我们就可以实现简单的python代码远程加载和更新功能.
总结


组件	作用
HookspecMarker("my_project")	定义 Hook 规范
HookimplMarker("my_project")	标记插件实现
PluginManager("my_project")	管理插件