嵌套字典与默认值字典合并

Question

我有 2 个嵌套词典，它们有一些匹配的键和类似的结构，并希望以特定的方式将它们合并到最后的第三个字典中。默认值字典具有将在第二个字典中使用的值(如果不在第二个字典中(，它将具有一些匹配的键和一些不存在的键。 无论哪种情况，我都希望它覆盖默认键或将第二个字典中的新键添加到第三个字典中。请参阅下面的(缩短(示例：

违约：

{"model_name": "null",
"description": "null",
"frequency": "d",
"tasks": [
{
"target": "elastic",
"metrics": "null",
"model_type": "null",
"alert": {
"type": "pagerduty",
"threshold": 5,
"service_id" : "P94CEA6"
}
}
]
}

第二句

{"model_name": "dqs_cie_registration_09",
"description": "test cie registration",
"tasks": [
{
"source": "elastic",
"metrics": [
"indid_unique_cnt", "zs"
],
"model_type": "Deep_Dive",
"elastic_config": "config",
"read_object": "dqs_rtfs_d_*",
"watcher": "cie_watch_zs_3d.json",
"target_write_index": "dqs_target_write_index"
}
]
}

我喜欢合并它，所以它会导致

{"model_name": "dqs_cie_registration_09",
"description": "test cie registration",
"frequency": "d",
"tasks": [
{
"target": "elastic",
"source": "elastic",
"metrics": ["indid_unique_cnt", "zs"],
"model_type": "Deep_Dive",
"elastic_config": "config",
"read_object": "dqs_rtfs_d_*",
"watcher": "cie_watch_zs_3d.json",
"target_write_index": "dqs_target_write_index",
"alert": {
"type": "pagerduty",
"threshold": 5,
"service_id" : "P94CEA6"
}
]
}

第三个词典将第二个词典合并到第一个词典上。

我还没有真正得到任何地方，但我觉得有一种非常简单的方法来实现这一点，我只是不记得了。

Answer 1

以下合并例程产生所需的结果

import copy    # to provide deepcopy
import pprint  # Pretty Print
def merge(a, b):
" Merges b into a (to preserve a make a deepcopy prior to calling merge "
if isinstance(a, dict) and isinstance(b, dict):
" Dictionaries "
for k, v in b.items():
if k in a:
# Conditionally add keys from b
if isinstance(a[k], str):
if a[k] == "null":
a[k] = copy.deepcopy(b[k])
else:
merge(a[k], b[k])
else:
# Add keys from b
a[k] = copy.deepcopy(b[k])
elif isinstance(a, list) and isinstance(b, list):
" Lists "
if len(a) == len(b):
for i, item in enumerate(b):
if isinstance(item, str) and isinstance(b[i], str):
if item == "null":
a[i] = b[i]
else:
merge(a[i], b[i])

用法

d1 = {"model_name": "null",
"description": "null",
"frequency": "d",
"tasks": [
{
"target": "elastic",
"metrics": "null",
"model_type": "null",
"alert": {
"type": "pagerduty",
"threshold": 5,
"service_id" : "P94CEA6"
}
}
]
}
d2 = {"model_name": "dqs_cie_registration_09",
"description": "test cie registration",
"tasks": [
{
"source": "elastic",
"metrics": [
"indid_unique_cnt", "zs"
],
"model_type": "Deep_Dive",
"elastic_config": "config",
"read_object": "dqs_rtfs_d_*",
"watcher": "cie_watch_zs_3d.json",
"target_write_index": "dqs_target_write_index"
}
]
}
merge(d1, d2) # to preserve d1 create a deepcopy prior to merge (i.e. temp = copy.deepcopy(d1))
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(d1)

输出

{   'description': 'test cie registration',    
'frequency': 'd',    
'model_name': 'dqs_cie_registration_09',    
'tasks': [   {   'alert': {   'service_id': 'P94CEA6', 
'threshold': 5,                       
'type': 'pagerduty'},
'elastic_config': 'config',
'metrics': ['indid_unique_cnt', 'zs'],
'model_type': 'Deep_Dive',
'read_object': 'dqs_rtfs_d_*',
'source': 'elastic',
'target': 'elastic',
'target_write_index': 'dqs_target_write_index',
'watcher': 'cie_watch_zs_3d.json'}
]
}