Python中實現簡單的插件框架

在系統設計中，常常咱們但願設計一套插件機制，在不修改程序主體狀況下，動態去加載附能。

我設想的插件系統：

一、經過類來實現
二、自動查找和導入

咱們假設須要實現一個簡單的插件系統，插件能夠接收一個參數執行。

實現基礎插件類

咱們先構建一個基礎插件類：plugin_collection.py

class Plugin:
    """
    該基類每一個插件都須要繼承，插件須要實現基類定義的方法"""
    def __init__(self):
        self.description = '未知'

    def perform_operation(self, argument):
        """
        實際執行插件所執行的方法，該方法全部插件類都須要實現
        """
        raise NotImplementedError

全部的插件類須要申明description來進行插件描述，而且要實現perform_operation方法，該方法是實際加載插件將去執行的方法。

簡易插件

咱們如今實現一個插件，實際執行時僅返回傳入的參數: plugins/identity.py

import plugin_collection

class Identity(plugin_collection.Plugin):
    """
    This plugin is just the identity function: it returns the argument
    """
    def __init__(self):
        super().__init__()
        self.description = 'Identity function'

    def perform_operation(self, argument):
        """
        The actual implementation of the identity plugin is to just return the
        argument
        """
        return argument

動態加載機制

由於咱們預實現動態加載插件。咱們經過定義一個PluginCollection來完成該職責，它將載入全部的插件，而且根據傳入的值執行perform_operation方法。PluginCollection類基礎組件實現以下：plugins_collection.py

class PluginCollection:
    """
    該類會經過傳入的package查找繼承了Plugin類的插件類
    """
    def __init__(self, plugin_package):
        self.plugin_package = plugin_package
        self.reload_plugins()

    def reload_plugins(self):
        """
        重置plugins列表，遍歷傳入的package查詢有效的插件
        """
        self.plugins = []
        self.seen_paths = []
        print()
        print(f"在 {self.plugin_package} 包裏查找插件")
        self.walk_package(self.plugin_package)

    def apply_all_plugins_on_value(self, argument):
        print()
        print(f"執行參數 {argument} 到全部的插件:")
        for plugin in self.plugins:
            print(f"    執行 {plugin.description} 參數 {argument} 結果 {plugin.perform_operation(argument)}")

最關鍵的是PluginCollection類裏的walk_package方法，該方法按以下步驟操做：

一、操做package裏全部的模塊
二、針對找到的模塊，檢查是不是Plugin的子類，非Plugin自身。每一個插件將會初始化並加入到列表。該檢查的好處是你能夠放入其餘Python模塊，也並不影響插件的使用
三、檢查當前package下的子目錄，遞歸查找插件

def walk_package(self, package):
        """
        遞歸遍歷包裏獲取全部的插件
        """
        imported_package = __import__(package, fromlist=['blah'])
        
        for _, pluginname, ispkg in pkgutil.iter_modules(imported_package.__path__, imported_package.__name__ + '.'):
            if not ispkg:
                plugin_module = __import__(pluginname, fromlist=['blah'])
                clsmembers = inspect.getmembers(plugin_module, inspect.isclass)
                for (_, c) in clsmembers:
                    # 僅加入Plugin類的子類，忽略掉Plugin自己
                    if issubclass(c, Plugin) and (c is not Plugin):
                        print(f'    找到插件類: {c.__module__}.{c.__name__}')
                        self.plugins.append(c())
    
        # 如今咱們已經查找了當前package中的全部模塊，如今咱們遞歸查找子packages裏的附件模塊
        all_current_paths = []
        if isinstance(imported_package.__path__, str):
            all_current_paths.append(imported_package.__path__)
        else:
            all_current_paths.extend([x for x in imported_package.__path__])
        
        for pkg_path in all_current_paths:
            if pkg_path not in self.seen_paths:
                self.seen_paths.append(pkg_path)
    
                # 獲取當前package中的子目錄
                child_pkgs = [p for p in os.listdir(pkg_path) if os.path.isdir(os.path.join(pkg_path, p))]
    
                # 遞歸遍歷子目錄的package
                for child_pkg in child_pkgs:
                    self.walk_package(package + '.' + child_pkg)

測試

如今咱們寫個簡單的測試:test.py

from plugin_collection import PluginCollection

my_plugins = PluginCollection('plugins')
my_plugins.apply_all_plugins_on_value(5)

執行結果：

$ python3 test.py 

在 plugins 包裏查找插件
    找到插件類: plugins.identity.Identity

執行參數 5 到全部的插件:
    執行 Identity function 參數 5 結果 5

代碼：https://github.com/erhuabushuo/plugin_template