Python使用pkgutil模块实现动态插件系统

pkgutil.iter_modules(path=None, prefix='')

# jobs/base.py
from typing import Protocol, runtime_checkable

@runtime_checkable
class JobProtocol(Protocol):
    """插件协议定义"""
    
    def enabled(self) -> bool:
        """判断插件是否启用"""
        ...
    
    def run(self) -> bool:
        """执行插件任务"""
        ...

# main.py
import logging
import pkgutil
import importlib
from types import ModuleType
from jobs.base import JobProtocol

logger = logging.getLogger(__name__)

def load_jobs(package: str = "jobs") -> list[JobProtocol]:
    """动态加载指定包下的所有任务插件"""
    loaded_jobs: list[JobProtocol] = []

    # 导入目标包
    pkg = importlib.import_module(package)
    
    # 遍历子包
    for finder, name, ispkg in pkgutil.iter_modules(pkg.__path__, pkg.__name__ + "."):
        # 只处理子包
        if not ispkg:
            continue
        
        # 动态导入模块
        module = importlib.import_module(name)
        
        # 获取工厂函数并创建实例
        if hasattr(module, "job_factory"):
            job = module.job_factory()
            
            # 协议验证
            if isinstance(job, JobProtocol) and job.enabled():
                loaded_jobs.append(job)
    
    return loaded_jobs

# jobs/jobs1/__init__.py
from jobs.base import JobProtocol
from .job import MyJob1

def job_factory() -> JobProtocol:
    return MyJob1()

# jobs/jobs1/job.py
class MyJob1:
    def enabled(self) -> bool:
        return True

    def run(self) -> bool:
        print(f"{self.__class__.__name__} is running")
        return True

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

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

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()

def run_jobs(jobs: list[JobProtocol]) -> None:
    for job in jobs:
        try:
            job.run()
        except Exception as e:
            logger.error(f"任务执行失败: {e}")
            # 继续执行其他任务

"""动态任务加载器

使用 pkgutil 模块动态发现和加载 jobs 包下的所有任务插件。
"""

import logging
import pkgutil
import importlib
from types import ModuleType
from jobs.base import JobProtocol

# 配置日志
logging.basicConfig(
    level=logging.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()