Вложенный словарь объединяется со значением по умолчанию dict - PullRequest
Новая
       2

Вложенный словарь объединяется со значением по умолчанию dict

0 голосов
/ 08 января 2020

У меня есть 2 вложенных словаря с соответствующими ключами и схожей структурой, и я хочу объединить их в окончательный третий словарь указанным способом c. Словарь значений по умолчанию содержит значения, которые будут использоваться, если не во втором словаре, который будет иметь некоторые совпадающие ключи и некоторые ключи, которые не существуют. В любом случае я хочу перезаписать ключ по умолчанию или добавить новый ключ из второго словаря в этот третий словарь. См. (Сокращенный) пример ниже:

по умолчанию: 

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

второй дикт

Третий диктат объединяет второй диктант с первым.

На самом деле я нигде не получил, но чувствую, что есть действительно простой способ реализовать это, что я просто не помню.

1 Ответ

0 голосов
/ 10 января 2020

Следующая процедура объединения дает желаемый результат 

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'}
    ]
}
...