包结构规范确保插件目录是规范的 Python 包每个插件包必须包含__init__.py文件父包jobs/也应包含__init__.py确保__path__属性正确设置虽然 Python 3.3 支持命名空间包无__init__.py但显式定义包结构更加健壮错误处理策略动态加载过程中存在多种潜在的失败点需要逐一处理try: pkg importlib.import_module(package) except ImportError as e: logger.error(f导入包失败: {e}) return [] for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, prefix): try: module importlib.import_module(name) except Exception as e: logger.error(f加载模块 {name} 失败: {e}) continue if not hasattr(module, job_factory): continue try: job module.job_factory() except Exception as e: logger.error(f实例化插件 {name} 失败: {e}) continue使用 logging 替代 print生产环境中应使用logging模块import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, ) logger logging.getLogger(__name__)这提供了日志级别控制、时间戳、输出重定向等关键能力。Protocol 运行时检查typing.Protocol配合runtime_checkable装饰器支持运行时类型检查from typing import Protocol, runtime_checkable runtime_checkable class JobProtocol(Protocol): def enabled(self) - bool: ... def run(self) - bool: ... # 检查实例是否满足协议 if isinstance(job, JobProtocol): job.run()注意运行时检查仅验证方法是否存在不验证方法签名。如果参数类型不匹配运行时仍会报错。插件隔离与依赖管理避免循环导入插件模块不应导入主程序模块延迟导入插件内部的重量级依赖应在run()方法中导入而非模块顶层异常隔离每个插件的执行应该相互独立一个插件失败不应影响其他插件def run_jobs(jobs: list[JobProtocol]) - None: for job in jobs: try: job.run() except Exception as e: logger.error(f任务执行失败: {e}) # 继续执行其他任务插件顺序控制如果插件执行顺序很重要可以考虑以下策略使用插件名称前缀排序如jobs/01_init/、jobs/02_process/在协议中添加priority()方法在插件元数据中定义依赖关系性能考量iter_modules()遍历文件系统频繁调用可能影响性能考虑在程序启动时一次性加载所有插件后续使用缓存的插件列表对于大量插件可以考虑延迟加载lazy loading模式安全性考虑动态加载代码存在潜在安全风险仅从可信路径加载插件在沙箱环境中运行不受信任的插件限制插件的文件系统和网络访问权限补充代码示例main.py动态任务加载器 使用 pkgutil 模块动态发现和加载 jobs 包下的所有任务插件。 import logging import pkgutil import importlib from types import ModuleType from jobs.base import JobProtocol # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, datefmt%Y-%m-%d %H:%M:%S, ) logger logging.getLogger(__name__) def load_jobs(package: str jobs) - list[JobProtocol]: 动态加载指定包下的所有任务插件。 遍历 package 下的所有子包尝试导入每个子包并调用其 job_factory 函数 创建任务实例。只有实现了 JobProtocol 协议且 enabled() 返回 True 的 任务才会被执行。 Args: package: 要扫描的包名默认为 jobs。 Returns: 成功加载的任务实例列表。 loaded_jobs: list[JobProtocol] [] try: pkg: ModuleType importlib.import_module(package) except ImportError as e: logger.error(f导入包 {package} 失败: {e}) return loaded_jobs # pkg.__path__ 可能是 None当 package 是命名空间包但没有子包时 if not hasattr(pkg, __path__) or pkg.__path__ is None: logger.warning(f包 {package} 没有 __path__ 属性无法遍历子模块) return loaded_jobs for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ .): # 只处理子包跳过模块文件 if not ispkg: logger.debug(f跳过模块 {name}只加载子包) continue try: module: ModuleType importlib.import_module(name) except Exception as e: logger.error(f加载任务模块 {name} 失败: {e}) continue if not hasattr(module, job_factory): logger.warning(f模块 {name} 没有 job_factory 函数跳过) continue try: job: JobProtocol module.job_factory() # 使用 Protocol 的运行时检查功能验证协议实现 if not isinstance(job, JobProtocol): logger.warning( f任务 {name} 未实现 JobProtocol 协议缺少 enabled 或 run 方法 ) continue if not job.enabled(): logger.info(f任务 {name} 已禁用跳过) continue loaded_jobs.append(job) logger.info(f成功加载任务: {name}) except Exception as e: logger.error(f创建任务实例 {name} 失败: {e}) continue return loaded_jobs def run_jobs(jobs: list[JobProtocol]) - None: 执行所有任务。 Args: jobs: 要执行的任务实例列表。 for job in jobs: try: result job.run() logger.info(f任务 {job.__class__.__name__} 执行完成结果: {result}) except Exception as e: logger.error(f任务 {job.__class__.__name__} 执行失败: {e}) def main() - None: 程序入口函数。 logger.info(开始加载任务...) jobs load_jobs() logger.info(f共加载 {len(jobs)} 个任务) logger.info(开始执行任务...) run_jobs(jobs) logger.info(所有任务执行完毕) if __name__ __main__: main()