Python解析Kafka YAML配置触发KeyError问题排查求助

问题概述 #

构建Kafka用户与主题访问权限审计报告时，原本正常运行的Python脚本突然抛出KeyError: 'topics'，报错位置为遍历YAML文件projects节点时访问each_project['topics']。

原代码 #

import yaml
import glob

yaml_file_names = glob.glob('/tmp/kafka-topics/topics/*.yaml')
file1 = open("Access_Audit_Report_testenv.txt", "w")
file1.close()
for each_yaml_file in yaml_file_names:
    with open(each_yaml_file) as f:
        document = yaml.safe_load(f)
        current_context = document["context"]
        for each_project in document['projects']:
            file1 = open("Access_Audit_Report_testenv.txt", "a")
            for each_topic in each_project['topics']:
                topic_name = '.'.join([current_context, each_project['name'], each_topic['name']])
                file1.write(topic_name+"
")
                file1.write('Consumers:\n')

                for each_entry in each_topic['consumers']:
                    file1.write(str(each_entry)+"
")

                file1.write(''+"
")
                file1.write('Producers:\n')

                for each_entry in each_topic['producers']:
                    file1.write(str(each_entry)+"
")

                file1.write(''+"
")
file1.close()

示例YAML文件 #

---
context: "building_administration"
projects:
  - name: "school"
    topics:
      - name: "publish"
        plan: "default"
        consumers:
          - principal: "User:consumer_planning_test"
        producers:
          - principal: "User:producer_maintenance_test"

  - name: "iot"
    topics:
      - name: "metrics.publish"
        plan: "default"
        consumers:
          - principal: "User:consumer_maintenance_test"
          - principal: "User:consumer_planning_test"
          
        producers:
          - principal: "User:producer_planner_test"
...

错误栈 #

Traceback (most recent call last):
  File "/tmp/kafka-audit-builder/build-audit-testenv.py", line 13, in <module>
    for each_topic in each_project['topics']:
KeyError: 'topics'

排查思路 #

1. 核心原因：当前遍历的某个YAML文件中，projects数组存在未包含topics字段的项目条目——可能是新增了不规范的YAML文件，或是现有文件被修改导致字段缺失。
2. 定位问题：在脚本中添加日志，输出当前处理的文件名和项目数据，快速找到异常条目。

修复方案 #

1. 定位问题文件 #

在原脚本的for each_project in document['projects']:下方添加日志输出：

print(f"Processing file: {each_yaml_file}, project: {each_project}")

运行脚本后，就能看到哪个文件的哪个项目缺少topics字段，针对性修复数据。

2. 脚本健壮性优化 #

修改脚本，通过dict.get()方法避免KeyError，同时跳过无topics的项目；另外优化文件操作逻辑，避免重复打开关闭文件：

import yaml
import glob

yaml_file_names = glob.glob('/tmp/kafka-topics/topics/*.yaml')

# 先清空目标文件
with open("Access_Audit_Report_testenv.txt", "w") as file1:
    pass

# 以追加模式打开文件，全程复用
with open("Access_Audit_Report_testenv.txt", "a") as file1:
    for each_yaml_file in yaml_file_names:
        with open(each_yaml_file) as f:
            document = yaml.safe_load(f)
            current_context = document["context"]
            for each_project in document['projects']:
                # 获取topics字段，不存在则返回空列表
                topics = each_project.get('topics', [])
                if not topics:
                    project_name = each_project.get('name', 'unnamed')
                    print(f"Skipping project {project_name} in {each_yaml_file} (no topics configured)")
                    continue
                
                for each_topic in topics:
                    topic_name = '.'.join([current_context, each_project['name'], each_topic['name']])
                    file1.write(f"{topic_name}\n")
                    file1.write('Consumers:\n')

                    for each_entry in each_topic['consumers']:
                        file1.write(f"{str(each_entry)}\n")

                    file1.write('\n')
                    file1.write('Producers:\n')

                    for each_entry in each_topic['producers']:
                        file1.write(f"{str(each_entry)}\n")

                    file1.write('\n')

验证结果 #

运行优化后的脚本：

• 自动跳过无topics的项目，不再抛出KeyError
• 控制台会输出跳过的项目信息，方便后续修复数据
• 生成的审计报告与期望输出一致

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