Python实现免显式导入的处理器注册模式方案咨询
我的应用中维护了持续扩充的Todo任务列表,每个任务完成时需要生成对应的专属文档。为了通用化所有任务的处理工作流,我最初编写了如下TodoDocumentCreator类:
class TodoDocumentCreator: def __init__(self): self.creators = {} def register_for_todo(self, todo_name): if todo_name in self.creators: raise CreatorMethodAlreadyExists(duty_slug) def wrapper(func): self.creators[todo_name] = func return func return wrapper def execute( self, todo: Todo, **kwargs ): method = self.creators[todo.name] return method(**kwargs) document_creator = TodoDocumentCreator()
设计初衷是为不同类型的Todo注册专属处理函数,Todo完成时调用document_creator.execute(),根据Todo名称匹配执行对应的文档生成函数。但实际编写注册代码时出现了问题:
@document_creator.register_for_duty(todo_name="stromsteueranmeldung") def fill_stromsteueranmeldung_for_company(todo: Todo): # 业务处理代码
我必须在TodoDocumentCreator所在模块显式导入所有被装饰的处理函数,否则装饰器不会执行、注册逻辑不会生效。这完全违背了注册模式的设计初衷——如果需要手动维护导入列表,和直接写冗长的switch分支做函数映射没有本质区别,完全没必要用注册模式。
我期望达成的效果:
- 特定Todo(例如
stromsteueranmeldung)相关的所有代码集中存放,和其他Todo的专属代码、Todo处理通用流程代码完全解耦 - 新增处理器时完全不需要修改通用结构的核心代码
当前实现达不到解耦效果,还存在明显bug,核心问题有三个:
- 装饰器执行依赖模块导入:Python的装饰器是在模块首次被导入时执行的,如果处理器模块从来没被任何代码import过,装饰器代码根本不会运行,处理函数自然不会注册到全局的
document_creator实例里,这就是必须手动在核心模块导入所有处理器的根本原因。 - 方法名、变量名笔误:类里定义的注册方法叫
register_for_todo,实际写装饰器时调用的是register_for_duty,运行时直接抛属性不存在错误;重复注册异常里引用的duty_slug变量没有定义,真触发重复注册场景时会先抛NameError。 - execute方法传参缺失:定义的处理函数需要接收
todo参数,但调用时只传了**kwargs,实际运行会报缺少位置参数的错误。
要实现真正零侵入的自动注册,不需要手动导入处理器,可以用自动模块扫描+约定优于配置的思路实现,步骤如下:
1. 固定处理器存放目录
首先约定所有Todo处理器都放在统一目录下,比如todo_handlers/,每个Todo类型的处理逻辑单独存为一个文件,目录结构示例:
your_project/ ├── core/ │ └── document_creator.py # 存放TodoDocumentCreator核心定义 └── todo_handlers/ ├── __init__.py ├── stromsteueranmeldung.py # 对应stromsteueranmeldung类型的Todo处理 └── other_todo_type.py # 其他类型Todo的处理
2. 修复原有bug,增加自动扫描能力
修正原有代码的错误,新增自动扫描指定目录下所有模块的方法,模块导入时装饰器会自动完成注册:
import importlib from pathlib import Path # 先定义自定义异常 class CreatorMethodAlreadyExists(Exception): pass class TodoDocumentCreator: def __init__(self): self.creators = {} def register_for_todo(self, todo_name): if todo_name in self.creators: raise CreatorMethodAlreadyExists(f"Todo类型{todo_name}的处理器已存在") def wrapper(func): self.creators[todo_name] = func return func return wrapper def autodiscover(self, handlers_dir_path: str, package_name: str): """ 自动扫描指定目录下的所有处理器模块并导入,触发装饰器注册 :param handlers_dir_path: 处理器目录的绝对路径 :param package_name: 处理器目录对应的Python包名 """ handlers_path = Path(handlers_dir_path) for file in handlers_path.glob("*.py"): # 跳过__init__.py等特殊文件 if file.name.startswith("_"): continue module_name = f"{package_name}.{file.stem}" # 导入模块,自动触发模块内装饰器执行完成注册 importlib.import_module(module_name) def execute(self, todo: Todo, **kwargs): method = self.creators.get(todo.name) if not method: raise ValueError(f"未找到Todo类型 {todo.name} 对应的文档生成处理器") return method(todo=todo, **kwargs) # 全局单例 document_creator = TodoDocumentCreator()
3. 项目启动时触发一次扫描
在项目启动入口(比如服务main函数、框架初始化钩子)只需要调用一次autodiscover方法,后续新增处理器只要把文件放到约定的todo_handlers目录下,不需要修改任何核心代码:
from core.document_creator import document_creator # 项目启动时执行一次即可 document_creator.autodiscover( handlers_dir_path="/path/to/your_project/todo_handlers", package_name="todo_handlers" )
4. 处理器编写完全解耦
每个处理器文件的代码不需要做任何特殊处理,只需要引入装饰器注册即可,不需要在核心模块手动导入:
# todo_handlers/stromsteueranmeldung.py from core.document_creator import document_creator from core.models import Todo @document_creator.register_for_todo(todo_name="stromsteueranmeldung") def fill_stromsteueranmeldung_for_company(todo: Todo, **kwargs): # 该Todo类型的专属处理逻辑,和其他代码完全隔离 pass
- 如果需要支持跨包的插件式处理器,可以用Python的entry points机制实现,比目录扫描更灵活,适合大型多模块项目;同项目内的业务逻辑用目录扫描足够简单轻量。
- 注册逻辑里可以增加函数签名校验,提前检查被装饰函数的参数是否符合要求,避免运行时传参错误。
- 可以给处理器增加优先级、通用前置/后置钩子等能力,适配更复杂的工作流场景。
内容的提问来源于stack exchange,提问作者Sven

