def adj_list_deltas_json()

in openr/py/openr/cli/utils/utils.py [0:0]

• 34 lines of code
• 12 McCabe index (conditional complexity)

def adj_list_deltas_json(adj_deltas_list, tags=None):
    """
    Parses a list of adjacency list deltas (from func find_adj_list_deltas),
    and returns the data as a json-formatted dict, and a status code.
        {
            tag-down: [nodes_down],
            tag-up: [nodes_up],
            tag-update: [
                {
                    "old_adj": old_adj,
                    "new_adj": new_adj
                }
            ]
        }

    @param adj_deltas_list: list<(changeType, oldAdjacency, newAdjacency)>
    @param tags: 3-tuple(string). a tuple of labels for
        (in old only, in new only, in both but different)
    """
    if not tags:
        tags = "NEIGHBOR_DOWN, NEIGHBOR_UP, NEIGHBOR_UPDATE"

    return_code = 0
    nodes_down = []
    nodes_up = []
    nodes_update = []

    for data in adj_deltas_list:
        old_adj = object_to_dict(data[1]) if data[1] else None
        new_adj = object_to_dict(data[2]) if data[2] else None

        if data[0] == tags[0]:
            assert new_adj is None
            nodes_down.append(old_adj)
            return_code = 1
        elif data[0] == tags[1]:
            assert old_adj is None
            nodes_up.append(new_adj)
            return_code = 1
        elif data[0] == tags[2]:
            assert old_adj is not None and new_adj is not None
            nodes_update.append({tags[0]: old_adj, tags[1]: new_adj})
            return_code = 1
        else:
            raise ValueError(
                'Unexpected change type "{}" in adjacency deltas list'.format(data[0])
            )

    deltas_json = {}

    if nodes_down:
        deltas_json.update({tags[0]: nodes_down})
    if nodes_up:
        deltas_json.update({tags[1]: nodes_up})
    if nodes_update:
        deltas_json.update({tags[2]: nodes_update})

    return deltas_json, return_code