如何在子类化Task时继承并扩展TaskStatus枚举的状态选项?
解决方案:动态生成子类专属状态枚举
核心思路
通过定义通用基础状态枚举,再利用Python的__init_subclass__钩子,让每个Task子类自动合并基础状态与自身新增的状态,生成专属的状态枚举。既避免了基类状态污染,又无需重复编写通用状态,子类开发者只需简单配置新增状态即可。
实现代码
from enum import Enum # 定义通用基础任务状态 class BaseTaskStatus(Enum): SUCCESS = "SUCCESS" FAILURE = "FAILURE" class Task: # 子类可通过重写此属性添加专属状态,格式为 {状态名: 状态值} _extra_statuses = {} def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs) # 合并基础状态与子类新增状态 status_items = [(member.name, member.value) for member in BaseTaskStatus] status_items.extend(cls._extra_statuses.items()) # 动态创建子类专属的状态枚举 cls.TaskStatus = Enum(f"{cls.__name__}Status", status_items) def parse_data(self, data): # 基类默认解析逻辑,子类可按需重写 self.task_status = self.TaskStatus.SUCCESS if data.get("result") == "ok" else self.TaskStatus.FAILURE # 带重试逻辑的子类示例 class TaskWithRetry(Task): # 仅需新增RETRY状态 _extra_statuses = {"RETRY": "RETRY"} def parse_data(self, data): if data.get("need_retry"): self.task_status = self.TaskStatus.RETRY else: super().parse_data(data) # 带暂停/恢复逻辑的子类示例 class TaskWithPause(Task): _extra_statuses = {"PAUSE": "PAUSE", "RESUME": "RESUME"} def parse_data(self, data): if data.get("is_paused"): self.task_status = self.TaskStatus.PAUSE elif data.get("is_resumed"): self.task_status = self.TaskStatus.RESUME else: super().parse_data(data)
方案优势
- 无状态污染:每个子类的
TaskStatus是独立枚举,新增状态不会影响其他Task子类。 - 无重复代码:自动继承
SUCCESS/FAILURE等通用状态,无需重复定义。 - 易扩展:子类开发者仅需修改
_extra_statuses字典即可添加新状态,学习成本极低。 - 类型安全:保留了Enum的类型校验特性,避免字符串硬编码的潜在错误。
- 兼容性好:完全兼容原有的Task类逻辑,子类重写
parse_data时可直接使用专属状态枚举。
内容的提问来源于stack exchange,提问作者Jagerber48
相关产品推荐
相关产品推荐

