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

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"
   }
  ]
 }

Id喜欢合并它，所以它导致

{"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'}
    ]
}