You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在子类化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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.15 11:55:26