如何通过MQTT主题拆分的键列表动态更新嵌套字典的值？

核心思路 #

针对MQTT动态主题路径更新字典的需求，需要实现任意长度键路径的自动遍历、中间键自动创建、值差异检查更新三个核心功能，同时保证代码简洁易维护。

简洁实现（循环版） #

import json

# 假设gen_dict是已加载的内存字典
# gen_dict = json.load(open('gen_dict.json', 'r'))

# 处理MQTT消息
topics = message.topic.split('/')
target_value = int(message.payload.decode())  # 根据实际payload格式解析，这里示例转为int
key_path = topics[1:]  # 移除根主题，保留字典的键路径

# 遍历键路径，创建缺失的中间字典
current_layer = gen_dict
for key in key_path[:-1]:
    if key not in current_layer:
        current_layer[key] = {}
    current_layer = current_layer[key]

# 检查并更新最后一个键的值
last_key = key_path[-1]
if current_layer.get(last_key) != target_value:
    current_layer[last_key] = target_value
    # 同步更新到文件
    with open('gen_dict.json', 'w') as f:
        json.dump(gen_dict, f, indent=2)

代码说明 #

• 动态键路径处理：通过key_path = topics[1:]提取根主题后的所有层级，适配任意长度的主题路径
• 自动创建中间键：循环遍历除最后一个键外的所有路径，若键不存在则创建空字典，确保层级可达
• 值更新判断：仅当目标值与当前值不同时才更新，避免无意义的字典修改
• 文件同步：更新内存字典后立即写入文件，保证内存与文件数据一致

递归实现（可选） #

如果偏好递归风格，也可以用以下方式实现：

def update_nested_dict(dictionary, keys, value):
    if len(keys) == 1:
        if dictionary.get(keys[0]) != value:
            dictionary[keys[0]] = value
            return True
        return False
    current_key = keys[0]
    if current_key not in dictionary:
        dictionary[current_key] = {}
    return update_nested_dict(dictionary[current_key], keys[1:], value)

# 使用示例
topics = message.topic.split('/')
target_value = int(message.payload.decode())
key_path = topics[1:]
if update_nested_dict(gen_dict, key_path, target_value):
    # 同步到文件
    with open('gen_dict.json', 'w') as f:
        json.dump(gen_dict, f, indent=2)

原代码问题分析 #

原代码存在两个致命问题：

1. 硬编码层级：仅能处理固定3层的主题路径，无法适配长度变化的主题
2. 异常处理无效：当中间键（如born）不存在时，gen_dict[topics[1]]会触发KeyError，此时except块中再次执行gen_dict[topics[1]][topics[2]][topics[3]] = 100依然会报错，因为中间层级未创建

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