Python3遍历字典键值对：Airflow DAG中SQL批量执行任务实现求助

问题场景 #

现有如下字典结构，键对应SQL文件名，值是需要执行该SQL的表名列表：

dict1 = {'a':['1','2','3'],'b':['1','2'],'c':['1']}

需要为每个SQL文件+表的组合生成对应的SnowflakeOperator任务，但原代码运行时出现重复task_id错误，无法正确实现循环逻辑。

原错误代码 #

dag_id = 'dag1'
@dag(
# 通用DAG参数在此处
)
def dag():
  for i in dict1: 
     for table in i:
        operator = SnowflakeOperator(
           task_id=f'{dag_id}_{table}_{i}',
          # 其他参数如下
        )

问题根源 #

原代码中for i in dict1遍历的是字典的键，后续for table in i实际是遍历键的字符（比如键为'a'时，会循环出字符'a'），既没有拿到真正的表名列表，还会生成重复的task_id。

正确实现代码 #

需要遍历字典的键值对，针对每个SQL文件对应的表列表生成唯一任务：

dag_id = 'dag1'
@dag(
# 通用DAG参数在此处
)
def dag():
    # 遍历字典键值对：sql_file是SQL文件名，tables是对应的表名列表
    for sql_file, tables in dict1.items():
        # 为每个表生成独立任务
        for table in tables:
            operator = SnowflakeOperator(
                task_id=f'{dag_id}_{sql_file}_{table}',
                # 关联对应SQL文件，示例：sql=f'/sql/scripts/{sql_file}.sql'
                # 补充其他必要参数：snowflake_conn_id、database等
            )

关键说明 #

• 使用dict1.items()直接获取键值对，确保拿到每个SQL文件对应的完整表列表
• task_id采用dag_id_sql_file_table格式，保证每个任务ID唯一，避免重复报错
• 可通过sql_file变量动态关联对应SQL文件路径，实现"指定SQL在指定表上执行"的需求

内容的提问来源于stack exchange，提问作者Potato Joe