Airflow子类Operator动态映射时如何强制设置max_active_tis_per_dagrun=1？

在动态任务映射（partial+expand）场景中，直接重写Operator的__init__方法无法生效，原因是partial()作为类方法会先创建模板实例，后续的任务实例生成逻辑不会再次触发__init__，而是依赖partial传入的参数进行初始化。

正确的实现方式有两种：

方式一：重写partial类方法 #

强制在调用partial时注入max_active_tis_per_dagrun=1参数，确保所有基于该模板生成的映射任务都继承这个配置：

class BetterKubernetesPodOperator(KubernetesPodOperator):
    @classmethod
    def partial(cls, **kwargs):
        # 强制覆盖参数，不管用户是否传入
        kwargs["max_active_tis_per_dagrun"] = 1
        return super().partial(**kwargs)

使用时直接调用partial+expand即可，无需额外传参：

op = BetterKubernetesPodOperator.partial(task_id="my_task").expand(...)

方式二：重写__init__并强制锁定属性 #

如果希望同时覆盖普通实例化和动态映射场景，可以在__init__中强制设置属性，同时重写__setattr__防止后续被修改：

class EvenBetterKubernetesPodOperator(KubernetesPodOperator):
    def __init__(self, **kwargs) -> None:
        # 强制传入参数到父类初始化
        kwargs["max_active_tis_per_dagrun"] = 1
        super().__init__(**kwargs)
    
    def __setattr__(self, name, value):
        # 锁定max_active_tis_per_dagrun属性，禁止后续修改
        if name == "max_active_tis_per_dagrun" and hasattr(self, name):
            return
        super().__setattr__(name, value)

这种方式既保证普通实例化时参数被强制设置，也能防止动态映射过程中参数被意外覆盖。

内容的提问来源于stack exchange，提问作者Kombajn zbożowy