I want to merge two nested objects so that the 2nd is layed out on top of the 1st. This is for a kubectl
config yaml but it could be any nested object that has a combination of list
s, dict
s and simple data types. eg:
# yaml 1:
containers:
- volumeMounts:
- name: external-stroage-1
mountPath: /mnt/stroage_1
readOnly: true
# Yaml 2
containers:
- name: cron
volumeMounts:
- name: internal-storage
mountPath: /mnt/data
And the merged object would be:
containers:
- name: cron
- volumeMounts:
- name: external-stroage-1
mountPath: /mnt/stroage_1
readOnly: true
- name: internal-storage
mountPath: /mnt/data
And here is what I have so far:
def merge(object_one, object_two):
assert type(object_one) == type(object_two), "Mismatched types"
if isinstance(object_one, dict):
for key in object_two:
if key in object_one:
object_one[key] = merge(object_one[key], object_two[key])
else:
object_one[key] = object_two[key]
elif isinstance(object_one, list):
for item in object_two:
object_one.append(item) # <<<<< when should I overwrite instead of append?
else:
return object_two
return object_one
Most of this can be done with simple recursion. Its easy to identify where an item should be inserted in a dict
since it's indexed by keys. But how do you identify if two items should be merged when you have a list
of objects (if the list order isn't guaranteed to be the same)? AKA, how do I determine if an item in a list needs to be overwritten vs appended? As it stands now, all list items are appended which leads to a bad merge:
containers:
- volumeMounts:
- name: external-stroage-1
mountPath: /mnt/stroage_1
readOnly: true
- name: external-stroage-2
mountPath: /mnt/stroage_2
- name: cron
volumeMounts: # This item should have been merged instead of being repeated
- name: internal-storage
mountPath: /mnt/data
This ended up being a bit more bloated than I was hoping, and has to make a big assumption about the yaml structure, which is to assume two dict
items within a list are the same item if they have a name
key and matching values (which is common in our kubectl yamls). But with that assumption, this did the job:
def merge(object_one, object_two):
"""
Recursively merge object_two over object one. This is not universal and makes some assumptions based on typical structure of kubectl/plato yamls
"""
assert type(object_one) == type(object_two), f"Mismatched types for object_one '{object_one}' and object_two {object_two}"
if isinstance(object_one, dict):
# If two dicts have a "name" field, and they match, its assumed they are the same field
if 'name' in object_one.keys() and 'name' in object_two.keys():
if object_one['name'] == object_two['name']:
return object_two
# Add missing keys to object_one
object_one = {**object_one, **{k: v for k, v in object_two.items() if k not in object_one}}
found = []
# If
for key in {x for x in object_one if x in object_two}:
if (tmp := merge(object_one[key], object_two[key])) is not None:
object_one[key] = tmp
found.append(True)
else:
found.append(False)
# If none is returned, the object is merged from 1 level up
if not all(found):
return None
elif isinstance(object_one, list):
# Compare every list element against the 2nd object, if no match is found, return None
for index_two in range(len(object_two)):
found = []
for index in range(len(object_one)):
try:
if tmp := merge(object_one[index], object_two[index_two]):
object_one[index] = tmp
found.append(True)
else:
found.append(False)
except Exception:
pass
# If None is returned, the object is merged from 1 level up
if not any(found):
object_one.append(object_two[index_two])
else:
# If both objects dont' match, return None which signals to the previous stack to merge one level up
if object_one == object_two:
return object_one
return
return object_one