插件系統讓應用程式可以在不修改核心程式碼的情況下擴展功能。本章介紹三種常見的插件架構模式，以及如何使用 Python 的動態載入機制實現它們。

先備知識

插件架構模式

模式一：基於繼承

最簡單的插件模式，插件必須繼承基類：

 1from abc import ABC, abstractmethod
 2from typing import ClassVar
 3
 4class Plugin(ABC):
 5    """插件基類"""
 6    name: ClassVar[str]  # 每個插件必須定義名稱
 7
 8    @abstractmethod
 9    def execute(self, context: dict) -> dict:
10        """執行插件邏輯"""
11        ...
12
13class LoggingPlugin(Plugin):
14    name = "logging"
15
16    def execute(self, context: dict) -> dict:
17        print(f"Processing: {context}")
18        return context
19
20class ValidationPlugin(Plugin):
21    name = "validation"
22
23    def execute(self, context: dict) -> dict:
24        if not context.get("valid"):
25            raise ValueError("Invalid context")
26        return context

優點：簡單明確，IDE 支援好 缺點：強制耦合，無法使用第三方類別

模式二：基於註冊

使用裝飾器或顯式註冊，更靈活：

 1from typing import Callable, TypeAlias
 2
 3PluginFunc: TypeAlias = Callable[[dict], dict]
 4
 5class PluginRegistry:
 6    """插件註冊表"""
 7
 8    def __init__(self) -> None:
 9        self._plugins: dict[str, PluginFunc] = {}
10
11    def register(self, name: str) -> Callable[[PluginFunc], PluginFunc]:
12        """裝飾器：註冊插件"""
13        def decorator(func: PluginFunc) -> PluginFunc:
14            if name in self._plugins:
15                raise ValueError(f"Plugin '{name}' already registered")
16            self._plugins[name] = func
17            return func
18        return decorator
19
20    def get(self, name: str) -> PluginFunc | None:
21        return self._plugins.get(name)
22
23    def list_plugins(self) -> list[str]:
24        return list(self._plugins.keys())
25
26# 全域註冊表
27registry = PluginRegistry()
28
29@registry.register("uppercase")
30def uppercase_plugin(context: dict) -> dict:
31    context["text"] = context.get("text", "").upper()
32    return context
33
34@registry.register("lowercase")
35def lowercase_plugin(context: dict) -> dict:
36    context["text"] = context.get("text", "").lower()
37    return context
38
39# 使用
40plugin = registry.get("uppercase")
41if plugin:
42    result = plugin({"text": "Hello"})

優點：靈活，支援函式和類別 缺點：需要顯式註冊

模式三：基於發現（使用 __init_subclass__）

利用 Python 的元編程機制自動發現插件：

 1from typing import ClassVar
 2
 3class AutoRegisterPlugin:
 4    """自動註冊的插件基類"""
 5    _registry: ClassVar[dict[str, type["AutoRegisterPlugin"]]] = {}
 6    name: ClassVar[str]
 7
 8    def __init_subclass__(cls, **kwargs) -> None:
 9        super().__init_subclass__(**kwargs)
10        # 跳過沒有定義 name 的中間類別
11        if hasattr(cls, "name") and cls.name:
12            if cls.name in AutoRegisterPlugin._registry:
13                raise ValueError(f"Plugin '{cls.name}' already registered")
14            AutoRegisterPlugin._registry[cls.name] = cls
15
16    @classmethod
17    def get_plugin(cls, name: str) -> type["AutoRegisterPlugin"] | None:
18        return cls._registry.get(name)
19
20    @classmethod
21    def list_plugins(cls) -> list[str]:
22        return list(cls._registry.keys())
23
24    def execute(self, context: dict) -> dict:
25        raise NotImplementedError
26
27# 定義插件只需要繼承，自動註冊
28class FormatPlugin(AutoRegisterPlugin):
29    name = "format"
30
31    def execute(self, context: dict) -> dict:
32        context["formatted"] = True
33        return context
34
35class CompressPlugin(AutoRegisterPlugin):
36    name = "compress"
37
38    def execute(self, context: dict) -> dict:
39        context["compressed"] = True
40        return context
41
42# 自動發現
43print(AutoRegisterPlugin.list_plugins())  # ['format', 'compress']

優點：無需顯式註冊，只要繼承就會被發現 缺點：必須繼承基類，Python 專屬

使用 Metaclass 的進階版本

 1from typing import ClassVar
 2
 3class PluginMeta(type):
 4    """插件元類別"""
 5    _registry: dict[str, type] = {}
 6
 7    def __new__(mcs, name: str, bases: tuple, namespace: dict) -> type:
 8        cls = super().__new__(mcs, name, bases, namespace)
 9
10        # 跳過基類
11        if bases and hasattr(cls, "plugin_name"):
12            plugin_name = cls.plugin_name
13            if plugin_name in mcs._registry:
14                raise ValueError(f"Plugin '{plugin_name}' already registered")
15            mcs._registry[plugin_name] = cls
16
17        return cls
18
19    @classmethod
20    def get_plugin(mcs, name: str) -> type | None:
21        return mcs._registry.get(name)
22
23class PluginBase(metaclass=PluginMeta):
24    """插件基類"""
25    plugin_name: ClassVar[str]
26
27    def run(self) -> None:
28        raise NotImplementedError
29
30class MyPlugin(PluginBase):
31    plugin_name = "my_plugin"
32
33    def run(self) -> None:
34        print("MyPlugin running")
35
36# 使用
37plugin_cls = PluginMeta.get_plugin("my_plugin")
38if plugin_cls:
39    plugin_cls().run()

動態載入模組

importlib 基礎

 1import importlib
 2from types import ModuleType
 3
 4def load_module(module_path: str) -> ModuleType:
 5    """動態載入模組"""
 6    return importlib.import_module(module_path)
 7
 8def load_class(module_path: str, class_name: str) -> type:
 9    """從模組載入類別"""
10    module = importlib.import_module(module_path)
11    return getattr(module, class_name)
12
13# 使用
14# 假設有 myapp/plugins/custom.py 定義了 CustomPlugin
15plugin_cls = load_class("myapp.plugins.custom", "CustomPlugin")
16plugin = plugin_cls()

從檔案路徑載入

 1import importlib.util
 2from pathlib import Path
 3from types import ModuleType
 4
 5def load_module_from_path(path: Path) -> ModuleType:
 6    """從檔案路徑載入模組"""
 7    spec = importlib.util.spec_from_file_location(
 8        path.stem,  # 模組名稱
 9        path        # 檔案路徑
10    )
11    if spec is None or spec.loader is None:
12        raise ImportError(f"Cannot load module from {path}")
13
14    module = importlib.util.module_from_spec(spec)
15    spec.loader.exec_module(module)
16    return module
17
18# 使用
19plugin_module = load_module_from_path(Path("./plugins/custom_plugin.py"))

importlib.metadata 的 entry_points（現代做法）

這是 Python 套件生態系統推薦的插件發現機制：

 1# 在 pyproject.toml 中定義 entry points
 2"""
 3[project.entry-points."myapp.plugins"]
 4format = "myapp_format_plugin:FormatPlugin"
 5compress = "myapp_compress_plugin:CompressPlugin"
 6"""
 7
 8from importlib.metadata import entry_points
 9
10def discover_plugins(group: str) -> dict[str, type]:
11    """發現已安裝套件中的插件"""
12    plugins = {}
13
14    # Python 3.10+
15    eps = entry_points(group=group)
16
17    for ep in eps:
18        try:
19            plugin_cls = ep.load()
20            plugins[ep.name] = plugin_cls
21        except Exception as e:
22            print(f"Failed to load plugin {ep.name}: {e}")
23
24    return plugins
25
26# 發現所有 myapp.plugins 群組的插件
27plugins = discover_plugins("myapp.plugins")
28for name, cls in plugins.items():
29    print(f"Found plugin: {name} -> {cls}")

優點：

  • 標準化的發現機制
  • 支援第三方套件提供插件
  • pip 安裝即可使用

實際範例一：Hook 系統的插件化改造

基於入門系列的 Hook 系統概念，設計一個插件化的 Hook 框架：

  1from abc import ABC, abstractmethod
  2from dataclasses import dataclass
  3from enum import Enum
  4from pathlib import Path
  5from typing import ClassVar
  6import importlib.util
  7
  8class HookEvent(Enum):
  9    PRE_TOOL_USE = "PreToolUse"
 10    POST_TOOL_USE = "PostToolUse"
 11    SESSION_START = "SessionStart"
 12    SESSION_END = "SessionEnd"
 13
 14@dataclass
 15class HookContext:
 16    """Hook 執行的上下文"""
 17    event: HookEvent
 18    tool_name: str | None = None
 19    input_data: dict | None = None
 20
 21@dataclass
 22class HookResult:
 23    """Hook 執行結果"""
 24    success: bool
 25    message: str
 26    modified_context: HookContext | None = None
 27
 28class HookPlugin(ABC):
 29    """Hook 插件基類"""
 30    _registry: ClassVar[dict[str, list[type["HookPlugin"]]]] = {}
 31
 32    # 子類別必須定義的屬性
 33    name: ClassVar[str]
 34    events: ClassVar[list[HookEvent]]
 35    priority: ClassVar[int] = 100  # 預設優先級
 36
 37    def __init_subclass__(cls, **kwargs) -> None:
 38        super().__init_subclass__(**kwargs)
 39        if hasattr(cls, "name") and hasattr(cls, "events"):
 40            for event in cls.events:
 41                event_name = event.value
 42                if event_name not in HookPlugin._registry:
 43                    HookPlugin._registry[event_name] = []
 44                HookPlugin._registry[event_name].append(cls)
 45                # 按優先級排序
 46                HookPlugin._registry[event_name].sort(
 47                    key=lambda c: c.priority
 48                )
 49
 50    @abstractmethod
 51    def execute(self, context: HookContext) -> HookResult:
 52        """執行 Hook 邏輯"""
 53        ...
 54
 55    @classmethod
 56    def get_hooks_for_event(cls, event: HookEvent) -> list[type["HookPlugin"]]:
 57        """取得特定事件的所有 Hook"""
 58        return cls._registry.get(event.value, [])
 59
 60# 定義具體的 Hook 插件
 61class SecurityCheckHook(HookPlugin):
 62    """安全檢查 Hook"""
 63    name = "security_check"
 64    events = [HookEvent.PRE_TOOL_USE]
 65    priority = 10  # 高優先級，先執行
 66
 67    DANGEROUS_PATTERNS = ["rm -rf", "DROP TABLE", "sudo"]
 68
 69    def execute(self, context: HookContext) -> HookResult:
 70        if context.input_data:
 71            command = context.input_data.get("command", "")
 72            for pattern in self.DANGEROUS_PATTERNS:
 73                if pattern in command:
 74                    return HookResult(
 75                        success=False,
 76                        message=f"Blocked dangerous pattern: {pattern}"
 77                    )
 78        return HookResult(success=True, message="Security check passed")
 79
 80class LoggingHook(HookPlugin):
 81    """日誌記錄 Hook"""
 82    name = "logging"
 83    events = [HookEvent.PRE_TOOL_USE, HookEvent.POST_TOOL_USE]
 84    priority = 1000  # 低優先級，最後執行
 85
 86    def execute(self, context: HookContext) -> HookResult:
 87        print(f"[{context.event.value}] Tool: {context.tool_name}")
 88        return HookResult(success=True, message="Logged")
 89
 90# Hook 執行器
 91class HookRunner:
 92    """執行 Hook 的管理器"""
 93
 94    def run_hooks(self, event: HookEvent, context: HookContext) -> list[HookResult]:
 95        """執行特定事件的所有 Hook"""
 96        results = []
 97        hook_classes = HookPlugin.get_hooks_for_event(event)
 98
 99        for hook_cls in hook_classes:
100            hook = hook_cls()
101            try:
102                result = hook.execute(context)
103                results.append(result)
104                # 如果 Hook 失敗且事件是 PRE，中斷執行
105                if not result.success and event == HookEvent.PRE_TOOL_USE:
106                    break
107            except Exception as e:
108                results.append(HookResult(
109                    success=False,
110                    message=f"Hook {hook_cls.name} failed: {e}"
111                ))
112
113        return results
114
115# 使用
116runner = HookRunner()
117context = HookContext(
118    event=HookEvent.PRE_TOOL_USE,
119    tool_name="Bash",
120    input_data={"command": "ls -la"}
121)
122results = runner.run_hooks(HookEvent.PRE_TOOL_USE, context)

實際範例二：通用插件框架設計

一個更通用的插件框架，支援動態載入和生命週期管理：

  1from abc import ABC, abstractmethod
  2from dataclasses import dataclass, field
  3from enum import Enum, auto
  4from pathlib import Path
  5from typing import Any
  6import importlib.util
  7
  8class PluginState(Enum):
  9    UNLOADED = auto()
 10    LOADED = auto()
 11    ACTIVE = auto()
 12    ERROR = auto()
 13
 14@dataclass
 15class PluginInfo:
 16    """插件元資訊"""
 17    name: str
 18    version: str
 19    description: str
 20    author: str = ""
 21    dependencies: list[str] = field(default_factory=list)
 22
 23class BasePlugin(ABC):
 24    """插件基類"""
 25
 26    @property
 27    @abstractmethod
 28    def info(self) -> PluginInfo:
 29        """返回插件資訊"""
 30        ...
 31
 32    def on_load(self) -> None:
 33        """插件載入時呼叫"""
 34        pass
 35
 36    def on_unload(self) -> None:
 37        """插件卸載時呼叫"""
 38        pass
 39
 40    def on_activate(self) -> None:
 41        """插件啟用時呼叫"""
 42        pass
 43
 44    def on_deactivate(self) -> None:
 45        """插件停用時呼叫"""
 46        pass
 47
 48@dataclass
 49class LoadedPlugin:
 50    """已載入的插件包裝"""
 51    plugin: BasePlugin
 52    state: PluginState = PluginState.UNLOADED
 53    error: str | None = None
 54
 55class PluginManager:
 56    """插件管理器"""
 57
 58    def __init__(self, plugin_dir: Path | None = None) -> None:
 59        self._plugins: dict[str, LoadedPlugin] = {}
 60        self._plugin_dir = plugin_dir
 61
 62    def load_plugin(self, path: Path) -> str:
 63        """從檔案載入插件"""
 64        try:
 65            # 動態載入模組
 66            spec = importlib.util.spec_from_file_location(
 67                path.stem, path
 68            )
 69            if spec is None or spec.loader is None:
 70                raise ImportError(f"Cannot load {path}")
 71
 72            module = importlib.util.module_from_spec(spec)
 73            spec.loader.exec_module(module)
 74
 75            # 尋找 BasePlugin 的子類別
 76            plugin_cls = None
 77            for attr_name in dir(module):
 78                attr = getattr(module, attr_name)
 79                if (isinstance(attr, type) and
 80                    issubclass(attr, BasePlugin) and
 81                    attr is not BasePlugin):
 82                    plugin_cls = attr
 83                    break
 84
 85            if plugin_cls is None:
 86                raise ImportError(f"No plugin class found in {path}")
 87
 88            # 實例化插件
 89            plugin = plugin_cls()
 90            plugin_name = plugin.info.name
 91
 92            # 檢查依賴
 93            for dep in plugin.info.dependencies:
 94                if dep not in self._plugins:
 95                    raise ImportError(f"Missing dependency: {dep}")
 96
 97            # 呼叫生命週期方法
 98            plugin.on_load()
 99
100            self._plugins[plugin_name] = LoadedPlugin(
101                plugin=plugin,
102                state=PluginState.LOADED
103            )
104
105            return plugin_name
106
107        except Exception as e:
108            raise ImportError(f"Failed to load plugin from {path}: {e}")
109
110    def unload_plugin(self, name: str) -> None:
111        """卸載插件"""
112        if name not in self._plugins:
113            raise KeyError(f"Plugin '{name}' not found")
114
115        loaded = self._plugins[name]
116
117        # 先停用
118        if loaded.state == PluginState.ACTIVE:
119            self.deactivate_plugin(name)
120
121        # 呼叫生命週期方法
122        loaded.plugin.on_unload()
123
124        del self._plugins[name]
125
126    def activate_plugin(self, name: str) -> None:
127        """啟用插件"""
128        if name not in self._plugins:
129            raise KeyError(f"Plugin '{name}' not found")
130
131        loaded = self._plugins[name]
132        if loaded.state != PluginState.LOADED:
133            raise RuntimeError(f"Plugin '{name}' is not in LOADED state")
134
135        try:
136            loaded.plugin.on_activate()
137            loaded.state = PluginState.ACTIVE
138        except Exception as e:
139            loaded.state = PluginState.ERROR
140            loaded.error = str(e)
141            raise
142
143    def deactivate_plugin(self, name: str) -> None:
144        """停用插件"""
145        if name not in self._plugins:
146            raise KeyError(f"Plugin '{name}' not found")
147
148        loaded = self._plugins[name]
149        if loaded.state != PluginState.ACTIVE:
150            return
151
152        loaded.plugin.on_deactivate()
153        loaded.state = PluginState.LOADED
154
155    def discover_plugins(self) -> list[Path]:
156        """發現插件目錄中的所有插件"""
157        if self._plugin_dir is None:
158            return []
159
160        return list(self._plugin_dir.glob("*.py"))
161
162    def load_all_plugins(self) -> dict[str, str | None]:
163        """載入所有發現的插件"""
164        results: dict[str, str | None] = {}
165        for path in self.discover_plugins():
166            try:
167                name = self.load_plugin(path)
168                results[name] = None
169            except Exception as e:
170                results[path.stem] = str(e)
171        return results
172
173    def get_plugin(self, name: str) -> BasePlugin | None:
174        """取得插件實例"""
175        loaded = self._plugins.get(name)
176        return loaded.plugin if loaded else None
177
178    def list_plugins(self) -> list[PluginInfo]:
179        """列出所有已載入的插件"""
180        return [lp.plugin.info for lp in self._plugins.values()]
181
182# 範例插件（放在獨立檔案中）
183class GreeterPlugin(BasePlugin):
184    """簡單的問候插件"""
185
186    @property
187    def info(self) -> PluginInfo:
188        return PluginInfo(
189            name="greeter",
190            version="1.0.0",
191            description="A simple greeting plugin"
192        )
193
194    def on_activate(self) -> None:
195        print("Greeter plugin activated!")
196
197    def greet(self, name: str) -> str:
198        return f"Hello, {name}!"
199
200# 使用
201manager = PluginManager(Path("./plugins"))
202manager.load_all_plugins()
203manager.activate_plugin("greeter")
204
205greeter = manager.get_plugin("greeter")
206if isinstance(greeter, GreeterPlugin):
207    print(greeter.greet("World"))

設計考量

安全性

 1# 限制插件可以存取的模組
 2import sys
 3from types import ModuleType
 4
 5class SandboxedImporter:
 6    """限制插件可以 import 的模組"""
 7    ALLOWED_MODULES = {"json", "re", "datetime", "typing"}
 8
 9    def find_module(self, name: str, path=None):
10        if name.split(".")[0] not in self.ALLOWED_MODULES:
11            return self  # 攔截
12        return None
13
14    def load_module(self, name: str) -> ModuleType:
15        raise ImportError(f"Module '{name}' is not allowed in plugins")
16
17# 使用時插入到 sys.meta_path
18# sys.meta_path.insert(0, SandboxedImporter())

版本相容性

 1from packaging import version
 2
 3def check_compatibility(
 4    plugin_info: PluginInfo,
 5    app_version: str
 6) -> bool:
 7    """檢查插件與應用程式版本相容性"""
 8    # 插件可以定義 min_app_version 屬性
 9    min_version = getattr(plugin_info, "min_app_version", "0.0.0")
10    return version.parse(app_version) >= version.parse(min_version)

小結

模式適用場景
基於繼承簡單場景，需要型別安全
基於註冊需要靈活性，支援函式插件
基於發現自動發現，減少樣板程式碼
entry_points套件生態系統，第三方插件

思考題

  1. 三種插件模式各有什麼優缺點？
  2. 如何實現插件之間的依賴管理？
  3. 如何確保插件的安全性？

上一章：3.5.3 進階上下文管理 下一章：3.5.5 設計模式整合案例