Python 插件化系统怎么设计?别把动态导入当成全部

一个插件更新后，主程序启动直接挂了：

ERROR plugin.load name=invoice_export
TypeError: InvoicePlugin.run() missing 1 required positional argument: 'context'

这种问题我见过几次。代码里用了 importlib，目录也能自动扫描，看起来已经“插件化”了，实际上只是把 import 写得更绕。

真正的插件系统，至少要把三件事拆开：主程序定义规则，插件实现规则，加载器负责发现和隔离。少一层，后面都容易乱。

假设现在有个数据处理平台，文件上传后可能执行字段清洗、订单校验、报表导出。最开始一般会写成这样：

defhandle(task_type: str, payload: dict) -> dict:
if task_type == "clean_customer":
return clean_customer(payload)
elif task_type == "check_order":
return check_order(payload)
elif task_type == "export_invoice":
return export_invoice(payload)

raise ValueError(f"unknown task: {task_type}")

业务只有三种时没什么问题。等分支涨到二十多个，再来一个临时定制需求，这个函数就不太敢动了。

我一般先把插件协议钉死，不急着写扫描目录。

from typing import Protocol, Any


classTaskPlugin(Protocol):
    name: str
    version: str

defvalidate(self, payload: dict[str, Any]) -> None:
        ...

defrun(self, payload: dict[str, Any]) -> dict[str, Any]:
        ...

这里用 Protocol，而不是强迫所有插件继承某个基类。插件作者只要提供这些字段和方法就行，耦合会轻一点。

比如订单校验插件可以这样写：

classOrderRiskPlugin:
    name = "order_risk"
    version = "1.2.0"

defvalidate(self, payload: dict) -> None:
ifnot payload.get("order_no"):
raise ValueError("order_no is required")

        amount = payload.get("amount")
ifnot isinstance(amount, int) or amount < 0:
raise ValueError("amount must be a non-negative integer")

defrun(self, payload: dict) -> dict:
        self.validate(payload)

        flags: list[str] = []
if payload["amount"] > 50_000:
            flags.append("large_amount")

return {
"order_no": payload["order_no"],
"risk_flags": flags,
        }

注意，插件不要直接碰主程序里的数据库连接、全局配置和内部 Service。很多插件系统最后拆不动，就是因为插件拿到了太多东西。

确实需要运行环境时，传一个收窄过的上下文：

from dataclasses import dataclass
from logging import Logger


@dataclass(frozen=True)
classPluginContext:
    logger: Logger
    tenant_id: str
    trace_id: str

别把整个 Flask、Django 或 FastAPI 的应用对象塞进去。那不叫上下文，叫把主系统钥匙交出去了。

插件怎么被发现

小项目可以扫描目录，但生产系统我更愿意用 Python 包的 Entry Point。插件可以独立打包、独立升级，主程序也不用知道它装在哪里。

插件项目的 pyproject.toml：

[project]
name = "acme-order-risk"
version = "1.2.0"

[project.entry-points."acme.task_plugins"]
order_risk = "order_risk.plugin:OrderRiskPlugin"

主程序只加载约定好的分组：

from importlib.metadata import entry_points


defdiscover_plugins() -> dict[str, object]:
    loaded: dict[str, object] = {}

for item in entry_points(group="acme.task_plugins"):
try:
            plugin_class = item.load()
            plugin = plugin_class()

            name = getattr(plugin, "name", "")
ifnot name:
raise RuntimeError("plugin name is empty")

if name in loaded:
raise RuntimeError(f"duplicated plugin: {name}")

            loaded[name] = plugin
except Exception:
            logger.exception(
"plugin load failed",
                extra={"entry_point": item.name},
            )

return loaded

这里有个细节：单个插件加载失败，不要默认把整个主程序带死。

除非它是系统启动必须依赖的核心插件，否则记录错误、标记不可用，主程序继续启动更合适。线上最怕的不是一个导出插件坏了，而是它坏了以后登录、查询、下单全没了。

插件管理器也别写得太花：

classPluginManager:
def__init__(self, plugins: dict[str, TaskPlugin]):
        self._plugins = plugins

defexecute(self, name: str, payload: dict) -> dict:
        plugin = self._plugins.get(name)
if plugin isNone:
raise LookupError(f"plugin not found: {name}")

        started = time.monotonic()

try:
            result = plugin.run(payload)
            logger.info(
"plugin executed",
                extra={
"plugin": name,
"version": plugin.version,
"cost_ms": int(
                        (time.monotonic() - started) * 1000
                    ),
                },
            )
return result
except Exception:
            logger.exception(
"plugin execution failed",
                extra={"plugin": name},
            )
raise

我会把插件名、版本、耗时和异常都打进日志。没有这些字段，出问题时只能看到一句 plugin execution failed，最后还得猜到底加载了哪个版本。

动态加载之后，还有几个坑

第一个是版本兼容。

主程序协议升级后，旧插件不一定能跑。可以让插件声明协议版本：

classOrderRiskPlugin:
    api_version = 2

加载时直接检查，不兼容就拒绝启用。别等调用到一半再报参数不匹配。

第二个是超时。

普通 Python 函数没法安全地强制中断。如果插件来自不受信任的团队，或者可能执行重计算，我不会把它直接跑在主进程里，而是放到独立进程甚至独立服务。线程池只能隔开调度，隔不开内存泄漏，也挡不住插件调用 os._exit()。

第三个是配置。

插件只读取自己的配置段：

plugins:
order_risk:
enabled:true
large_amount:50000

不要让插件随便读取整份配置，更不要把数据库密码、签名密钥一起传进去。

插件化不是为了展示 Python 能动态导入模块。它解决的是另一件事：新能力进来时，主程序尽量不改；插件出问题时，故障范围尽量收住。

接口要窄，加载要有日志，版本要能拒绝，执行要考虑隔离。把这几处守住，插件系统才算能长期用。否则目录里多几个 .py 文件，离插件化还差得远。