Airflow Jinja模板无法与自定义操作员一起使用

Question

我正在尝试使自定义Airflow操作员如下所示：

class NewCheckOperator(BaseOperator):

    __mapper_args__ = {
        'polymorphic_identity': 'NewCheckOperator'
    }
    template_fields = ('sql1', 'sql2')  # type: Iterable[str]
    template_ext = ('.hql', '.sql',)  # type: Iterable[str]
    ui_color = '#91E6F2'

    ratio_formulas = {
        'max_over_min': lambda cur, ref: float(max(cur, ref)) / min(cur, ref),
        'relative_diff': lambda cur, ref: float(abs(cur - ref)) / ref,
    }

    @apply_defaults
    def __init__(
        self,
        table: str,
        template_time: str,
        metrics_thresholds: Dict[str, int],
        random_param: int, 
        ratio_formula: Optional[str] = 'max_over_min',
        ignore_zero: Optional[bool] = True,
        conn_id: Optional[str] = None,
        *args, **kwargs
    ):
        super().__init__(*args, **kwargs)
        #some logic here 

        self.sql1 = " '{{ ds }}' "
        self.sql2 = "'{{ macros.ds_add(ds, " + str(self.random_param) + ") }}'"

    def execute(self, context=None):
        hook = self.get_db_hook()
        self.log.info('Using ratio formula: %s', self.ratio_formula)
        self.log.info('Executing SQL check: %s', self.sql2)
        row2 = hook.get_first(self.sql2)
        self.log.info('Executing SQL check: %s', self.sql1)
        row1 = hook.get_first(self.sql1)

        if not row2:
            raise AirflowException("The query {} returned None".format(self.sql2))
        if not row1:
            raise AirflowException("The query {} returned None".format(self.sql1))

       #other logic 

    def get_db_hook(self):
        return BaseHook.get_hook(conn_id=self.conn_id)

class BigQueryNewCheckOperator(NewCheckOperator): 
    template_fields = ('table', 'gcp_conn_id', )

    @apply_defaults
    def __init__(self,
                 table: str,
                 metrics_thresholds: dict,
                 random_param: int,
                 date_filter_column: str = 'ds',                 
                 gcp_conn_id: str = 'google_cloud_default',
                 bigquery_conn_id: Optional[str] = None,
                 use_legacy_sql: bool = True,
                 *args,
                 **kwargs) -> None:
        super().__init__(
            table=table, metrics_thresholds=metrics_thresholds,
            date_filter_column=date_filter_column, days_back=days_back,
            *args, **kwargs)

        if bigquery_conn_id:
            warnings.warn(
                "The bigquery_conn_id parameter has been deprecated. You should pass "
                "the gcp_conn_id parameter.", DeprecationWarning, stacklevel=3)
            gcp_conn_id = bigquery_conn_id

        self.gcp_conn_id = gcp_conn_id
        self.use_legacy_sql = use_legacy_sql

    def get_db_hook(self):
        return BigQueryHook(bigquery_conn_id=self.gcp_conn_id,
                            use_legacy_sql=self.use_legacy_sql)

但是由于某些原因，NewCheckOperator中的模板不起作用（通过这种方式，我认为它们像str一样被视为" '{{ ds }}' "文字），尽管将sql1和sql2显式设置为template_fields 。这是在IntervalCheckOperator和BigQueryIntervalCheckOperator之后紧密建模的，因此我不确定为什么气流宏无法正常工作。在这里的任何帮助将不胜感激！