Python结合重载与策略模式实现符合开闭原则的邮件分发方案咨询
无硬编码分支的枚举策略模式实现方案
核心思路是通过自动注册的策略映射实现分发,完全避免针对枚举值的多重if/switch业务分支,新增接收者类型时无需修改原有分发逻辑,完全符合开闭原则,同时自带强制类型校验能力。
方案1:装饰器驱动的独立策略注册表
这个方案实现最灵活,策略逻辑和枚举定义解耦,适合策略逻辑比较复杂的场景。
首先实现通用的分发器和注册装饰器:
import enum from typing import Callable, Dict # 原有接收者枚举定义保持不变 class Receiver(enum.Enum): ALL = "ALL" THREE_MONTHS = "THREE_MONTHS" SIX_MONTHS = "SIX_MONTHS" # 全局策略映射表,自动维护枚举值到对应处理函数的关系 _strategy_map: Dict[Receiver, Callable] = {} def register_strategy(receiver_type: Receiver): """装饰器:注册指定接收者类型对应的邮件发送策略""" def wrapper(func: Callable): if receiver_type in _strategy_map: raise ValueError(f"接收者类型 {receiver_type} 的处理策略已重复注册") _strategy_map[receiver_type] = func return func return wrapper def send_to(receiver_type: Receiver): """统一分发入口,无业务分支判断""" # 以下为边界合法性校验,不属于业务分支,新增枚举值时无需修改 if not isinstance(receiver_type, Receiver): raise TypeError(f"参数必须为Receiver枚举类型,非法传入类型:{type(receiver_type)}") try: handler = _strategy_map[receiver_type] except KeyError: raise NotImplementedError(f"接收者类型 {receiver_type} 未实现对应处理逻辑") return handler()
对应不同接收者的处理逻辑,按期望的和枚举值绑定的形式编写即可,新增逻辑只需要加新的被装饰函数:
@register_strategy(Receiver.ALL) def _(): # 全体用户发送逻辑 print("执行全体用户邮件投递") @register_strategy(Receiver.THREE_MONTHS) def _(): # 近3个月登录用户发送逻辑 print("执行近3个月登录用户邮件投递") @register_strategy(Receiver.SIX_MONTHS) def _(): # 近6个月登录用户发送逻辑 print("执行近6个月登录用户邮件投递")
上层接口逻辑只需要补充字符串转枚举的参数校验即可,原有调用逻辑无需改动:
@api_view(["POST"]) def send_email(request): email_to_str = request.data["emailTo"] # 校验入参合法性,将前端传入的字符串转为枚举实例 try: email_to = Receiver(email_to_str) except ValueError: return Response({"code": 400, "msg": "非法的接收者类型"}, status=400) send_to(email_to) return Response({"code": 200, "msg": "邮件投递任务已提交"})
这个方案的优势:
- 完全符合开闭原则:后续新增接收人群类型,只需要在
Receiver枚举中新增成员,再写一个带@register_strategy(新枚举成员)装饰器的处理函数即可,不需要修改send_to分发入口、不需要改动任何已有业务代码 - 自带强制校验:非法类型参数、未注册策略的枚举值都会抛出明确的错误
- 策略逻辑和枚举、分发逻辑完全解耦,方便拆分不同策略到独立模块维护
方案2:枚举类内置策略绑定
如果希望策略和枚举定义收敛在一起,可以直接在枚举类中实现注册和分发能力,不需要额外维护全局映射表:
import enum from typing import Callable class Receiver(enum.Enum): ALL = "ALL" THREE_MONTHS = "THREE_MONTHS" SIX_MONTHS = "SIX_MONTHS" def __init__(self, value: str): self._handler: Callable = None def register_handler(self, func: Callable): """注册当前枚举值对应的处理函数""" if self._handler is not None: raise ValueError(f"{self.name} 类型的处理逻辑已重复注册") self._handler = func return func def handle(self, *args, **kwargs): """执行当前枚举值对应的处理逻辑""" if self._handler is None: raise NotImplementedError(f"{self.name} 类型未实现对应处理逻辑") return self._handler(*args, **kwargs)
对应的策略注册形式更简洁:
@Receiver.ALL.register_handler def send_all(): print("执行全体用户邮件投递") @Receiver.THREE_MONTHS.register_handler def send_three_months(): print("执行近3个月登录用户邮件投递") @Receiver.SIX_MONTHS.register_handler def send_six_months(): print("执行近6个月登录用户邮件投递")
统一分发入口可以进一步简化:
def send_to(receiver_type: Receiver): # 边界合法性校验,新增枚举值时无需修改 if not isinstance(receiver_type, Receiver): raise TypeError("参数必须为Receiver枚举类型") receiver_type.handle()
注意事项
Python标准库的functools.singledispatch是基于参数类型做分发,无法实现基于枚举成员(值)的路由效果,不适合当前场景。
以上两种方案的分发效果和Java中基于枚举参数的重载、多态分发效果完全一致,上层调用方只需要传入枚举实例即可自动匹配对应处理逻辑,不需要感知底层策略实现。
内容的提问来源于stack exchange,提问作者user19497472
相关产品推荐
相关产品推荐

