Python 插件化系统怎么设计？别把动态导入当成全部
python约 4 分钟一个插件更新后，主程序启动直接挂了：
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, AnyclassTaskPlugin(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_pointsdefdiscover_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 = pluginsdefexecute(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 文件，离插件化还差得远。