pkgutil 簡介
pkgutil 是 Python 標準庫中的一個模塊，提供了用於處理 Python 包的工具函數。它的核心功能之一是 iter_modules() 函數，能夠動態遍歷和發現指定包路徑下的所有子模塊和子包。這一特性使其成為實現動態插件系統的選擇之一。（之前也介紹過藉助__init_subclass__()在子類繼承時動態註冊插件）
與手動遍歷文件系統或使用第三方庫相比，pkgutil 具有以下優勢：
- 標準庫原生支持：無需引入額外依賴
- 跨平台兼容：統一處理不同操作系統的路徑差異
- 支持命名空間包：能夠正確處理 PEP 420 定義的命名空間包
- 與導入系統緊密集成：返回的模塊名可直接用於
importlib.import_module()
核心函數：iter_modules
iter_modules() 函數簽名如下：
pkgutil.iter_modules(path=None, prefix='')
- path：要搜索的路徑列表，通常使用包的
__path__屬性 - prefix：返回的模塊名前綴，常用於構建完整的模塊導入路徑
該函數返回一個迭代器，每個元素是一個三元組 (module_info_finder, name, ispkg)：
module_info_finder：查找器對象（Python 3.6+ 為ModuleInfo實例）name：模塊或包的名稱（不含前綴）ispkg：布爾值，表示是否為包（含有__init__.py的目錄）
實現動態插件系統
設計思路
一個典型的插件系統包含以下組件：
- 協議定義：使用
typing.Protocol定義插件必須實現的接口 - 插件發現：使用
pkgutil.iter_modules()自動發現所有插件包 - 插件加載：使用
importlib.import_module()動態導入插件模塊 - 插件驗證：運行時檢查插件是否滿足協議要求
- 插件執行：調用插件方法執行具體任務
代碼實現
首先定義插件協議：
# jobs/base.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class JobProtocol(Protocol):
"""插件協議定義"""
def enabled(self) -> bool:
"""判斷插件是否啓用"""
...
def run(self) -> bool:
"""執行插件任務"""
...
@runtime_checkable 裝飾器使得協議可以在運行時通過 isinstance() 進行檢查。
寫稿的時候想起來可以加個
_runable()方法， 執行run()方法之前先檢查是否滿足可執行條件。
插件加載器實現：
# 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
之所以每個插件放單獨的package中，是想着如果插件功能複雜，單個文件的篇幅可能會極長，可以拆分到不同的文件中。每個插件也可以維護單獨的配置加載方式。而且可以利用上 pkgutil 返回的
ispkg。如果插件的功能簡單，也可以寫成單獨的文件。
項目結構
project/
├── main.py # 主程序入口
├── jobs/
│ ├── __init__.py # 包初始化文件
│ ├── base.py # 協議定義
│ ├── jobs1/ # 插件1
│ │ ├── __init__.py # 導出 job_factory
│ │ └── job.py # 具體實現
│ ├── jobs2/ # 插件2
│ │ ├── __init__.py
│ │ └── job.py
│ └── jobs3/ # 插件3（可禁用）
│ ├── __init__.py
│ └── job.py
運行結果示例
2026-03-01 21:13:26 - INFO - 開始加載任務...
2026-03-01 21:13:26 - INFO - 成功加載任務: jobs.jobs1
2026-03-01 21:13:26 - INFO - 成功加載任務: jobs.jobs2
2026-03-01 21:13:26 - INFO - 任務 jobs.jobs3 已禁用，跳過
2026-03-01 21:13:26 - WARNING - 任務 jobs.jobs4 未實現 JobProtocol 協議（缺少 enabled 或 run 方法）
2026-03-01 21:13:26 - INFO - 共加載 2 個任務
2026-03-01 21:13:26 - INFO - 開始執行任務...
MyJob1 is running
2026-03-01 21:13:26 - INFO - 任務 MyJob1 執行完成，結果: True
MyJob2 is running
2026-03-01 21:13:26 - INFO - 任務 MyJob2 執行完成，結果: True
2026-03-01 21:13:26 - INFO - 所有任務執行完畢
實際應用注意事項
包結構規範
確保插件目錄是規範的 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(
level=logging.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(
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()
