基于两键组合查询MongoDB关联值时触发KeyError的问题

所有触发问题的点都是代码编写时的变量名串改、逻辑错位，无复杂环境问题：

• 变量名完全错配：代码中引用的param_type、param_value、distributions_per_type三个变量，从未在当前函数作用域内定义，和实际初始化的distros_per_param、赋值的param_one/param_two完全不匹配。当代码执行到往distributions_per_param[param_type, param_value]写入值的逻辑时，因为这个元组键从未提前初始化对应的集合结构，直接触发KeyError，抛出的错误信息里的元组值就是当时读取到的字段内容，也就是你看到的KeyError: ('test1', 'test2')。
• 分组逻辑和需求错位：原代码单独把param_one、param_two拆成两个顶层字典键存储，没有实现“两字段值拼接为一个组合键”的要求，和预期输出格式完全不匹配。
• 示例数据笔误：给出的测试样例里两个键都写成了param_one，实际MongoDB文档结构是param_one、param_two两个独立字段，后续读取param_two的时候也会触发KeyError。
• 计算逻辑位置错误：把结果字典的生成语句写在了循环内部，每遍历一条文档就重算一次全量结果，不仅浪费性能，还容易因为变量错位引入新的bug。
• 空值容错缺失：直接通过字典键访问文档字段，没有判断字段是否存在、值是否为空，只要文档缺任意一个字段就会抛错。

采用MongoDB原生聚合管道做服务端分组，比全量拉取文档到本地循环的性能高几个数量级，同时修正所有变量错配的问题，严格匹配预期输出格式。

from collections import OrderedDict
from typing import Dict, List
import pymongo
from tqdm import tqdm

def distros(
        mongo_uri: str,
        mongo_source_db: str,
        mongo_source_collection: str) -> Dict[str, List[str]]:
    distros_per_param = OrderedDict()
    
    mongo = pymongo.MongoClient(mongo_uri)
    db = mongo.get_database(mongo_source_db)
    col = db.get_collection(mongo_source_collection)
    
    # 聚合管道实现服务端分组
    pipeline = [
        # 过滤两个分组键、目标键存在且非空的文档
        {"$match": {
            "param_one": {"$exists": True, "$ne": ""},
            "param_two": {"$exists": True, "$ne": ""},
            "distro": {"$exists": True, "$ne": ""}
        }},
        # 按两个字段的组合分组
        {"$group": {
            "_id": {
                "p1": "$param_one",
                "p2": "$param_two"
            },
            # 自动去重收集同组的distro值
            "values": {"$addToSet": "$distro"}
        }}
    ]
    total = col.count_documents(pipeline[0]["$match"])
    cursor = col.aggregate(pipeline)
    
    for group_doc in tqdm(cursor, total=total, desc='distributions'):
        # 拼接成要求格式的组合键
        group_key = f"{group_doc['_id']['p1']}, {group_doc['_id']['p2']}"
        # 统一清洗distro值：去首尾空格、转小写、过滤空值
        cleaned_values = [
            str(v).strip().lower() 
            for v in group_doc['values'] 
            if str(v).strip()
        ]
        distros_per_param[group_key] = sorted(cleaned_values)
    
    # 如果需要列表式返回结构，可改为：return [ {k:v} for k,v in distros_per_param.items() ]
    return dict(distros_per_param)

• 自动过滤两个分组键、目标键为空/不存在的文档，避免字段缺失触发的KeyError
• 同分组下的distro值自动去重，统一完成格式清洗
• 组合键严格按"param_one值, param_two值"的格式拼接
• 同组的distro值自动按字典序排序
• 彻底消除原代码的变量错配问题，不会再抛出元组形式的KeyError

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