[Python] Airflow-Using for loop with 2 variable inputs

The above is how I have defined the input values in my config yaml file.

with TaskGroup(group_id='full_data_refresh') as full_data_refresh:

for item in itertools.product(dag_config['src_config_id'],dag_config['ctry_config_id']):

config_run_ids = f'{item}'

exec_funs_mti_all_functions_run = PostgresOperator(
task_id= f'exec_funs_mti_all_functions_run{config_run_ids}',
postgres_conn_id='DDL_CONN_WRT_CSB',
sql=f'select funs_mti_all_functions_run{config_run_ids}';'
)


This is the code I have used to run my sql function: funs_mti_all_functions_run(src_config_id text,ctry_config_id text) So, I want the task_id : f'exec_funs_mti_all_functions_run_{config_run_ids}' to run for every combination of src_config_id and ctry_config_id which is:

select funs_mti_all_functions_run('1','1');
select funs_mti_all_functions_run('1','2');
select funs_mti_all_functions_run('1','6');
select funs_mti_all_functions_run('1','10');


The implementation of the python file for the DAG definition although errors out with the message:

"airflow.exceptions.AirflowException: The key "exec_funs_mti_all_functions_run('1', '1')" has to be made of alphanumeric characters, dashes, dots and underscores exclusively"
​

Can someone help here?

Continue reading...