In Airflow, cannot check for equality using config variable passed

Canovice 0
Viewed 22

Our DAG

import func1
import func2

all_tables = [
    'tablea',
    'tableb',
    'tablec', 
    'tabled'
]

with TABLE_MONGO_LOAD:

    # Pass "table" parameter 
    this_table = "{{ dag_run.conf['table'] }}"

    this_function = func1 if this_table in all_tables else func2
    get_local_json = PythonOperator(
        task_id=f'bq_to_local',
        python_callable=this_function,
        op_kwargs={ 'table_name': this_table, 'file_name': this_table 
    )
   
    ...

We trigger this DAG manually, and pass { "table": "tablea" }. We want this_function = func1 if this_table in all_tables else func2 to lead to this_function equalling func, because tablea is in fact in all_tables, however everytime we run this DAG, func2 is returned. No matter what we pass to the config, this_function = func1 if this_table in all_tables else func2 always returns the else case.

Is this_table = "{{ dag_run.conf['table'] }}" not returning a string? Even when I do a super basic check:

if this_table == 'tablea':
    print('a')
else:
    print('b')

this always prints b even if the config passed is { "table": "tablea" }. What's going on here?

airflow
1 Answers

the line "this_function = func1 if this_table in all_tables else func2" is running before the dag is running while airflow is interpreting that the dag is valid.

in order to check while running the DAG you should use BranchPythonOperator

ozs0
Related