Airflow自定义可参数化Timetable时serialize方法报TypeError

错误原因 #

1. 类型注解语法不兼容：你的Airflow运行环境Python版本低于3.9，不支持dict[str, Any]这种直接对内置类型做下标注解的语法，此时dict作为类型对象无法被下标，触发TypeError: 'type' object is not subscriptable。
2. 属性访问错误：next_dagrun_info方法中使用了self.hour和self.minute，但类初始化时定义的私有属性是self._hour和self._minute，后续运行会触发AttributeError。

解决方案 #

方案1：适配Python <3.9的类型注解 #

1. 在文件顶部导入typing模块的对应类型：

from typing import Dict, Any, Optional

2. 修改serialize和deserialize方法的类型注解：

def serialize(self) -> Dict[str, Any]:
    return {"hour": self._hour.isoformat(), "minute": self._minute.isoformat()}

@classmethod
def deserialize(cls, value: Dict[str, Any]) -> Timetable:
    return cls(Time.fromisoformat(value["hour"]), Time.fromisoformat(value["minute"]))

方案2：使用future注解延迟解析 #

在文件最顶部添加以下导入，让Python将类型注解当作字符串处理，避免运行时解析错误：

from __future__ import annotations

修复属性访问错误 #

修改next_dagrun_info中的属性引用：

run_after=DateTime.combine(next_end.date(), self._hour, self._minute).replace(tzinfo=UTC),

内容的提问来源于stack exchange，提问作者Luis Lema