Airflow Template_fields added but variable like {{ ds }} is not working

Viewed 1074

I want to pass airflow variables to SQL query template file like this (in sql/test.sql file):

select 'test', '{{ params.test_ds }}', '{{ test_dt }}' from test_table;

I created a Operator inherited from PostgresOperator:

class EtlOperator(PostgresOperator):
    template_fields = ('sql', 'test_dt', 'params')
    template_ext = PostgresOperator.template_ext

    @apply_defaults
    def __init__(self, test_dt, params, *args, **kwargs):
        super(EtlRunIdOperator, self).__init__(*args, **kwargs)
        self.test_dt = test_dt
        self.params = params

    def execute(self, context):
        super(EtlRunIdOperator, self).execute(context)

I created this task:

test_task00 = EtlOperator(
    task_id=f'test_task00',
    postgres_conn_id='redshift',
    sql='sql/test.sql',
    params={
        'test_ds': '{{ ds }}'
    },
    database='default',
    test_dt='{{ execution_date }}',
    provide_context=True, # tried without it too
    dag=dag
)

However, no matter the params or test_dt that are part of template_fields, the SQL still not parsing the variables, results like this:

INFO - Executing: select 'test', '{{ ds }}', '' from test_table;

Is there anywhere wrong in my configurations?

0 Answers
Related