Search code examples
pythonpython-3.xlistdictionarypyyaml

Merge two nested lists of objects


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 lists, dicts 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

Solution

  • 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