Source code for json_merger.merger

# -*- coding: utf-8 -*-
# This file is part of Inspirehep.
# Copyright (C) 2016 CERN.
# Inspirehep is free software; you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
# Inspirehep is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with Inspirehep; if not, write to the
# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
# MA 02111-1307, USA.
# In applying this license, CERN does not
# waive the privileges and immunities granted to it by virtue of its status
# as an Intergovernmental Organization or submit itself to any jurisdiction.

"""Definition for JSON merger class."""

from __future__ import absolute_import, print_function

import copy

from .comparator import DefaultComparator
from .dict_merger import SkipListsMerger
from .errors import MergeError
from .list_unify import ListUnifier
from .utils import (
    get_conf_set_for_key_path, get_dotted_key_path,
    get_obj_at_key_path, set_obj_at_key_path


[docs]class Merger(object): """Class that merges two JSON objects that share a common ancestor. This class treats by default all lists as being lists of entities and offers support for matching their elements by their content, by specifing per-field comparator classes. """ def __init__(self, root, head, update, default_dict_merge_op, default_list_merge_op, list_dict_ops=None, list_merge_ops=None, comparators=None, data_lists=None): """ Args: root: A common ancestor of the two objects being merged. head: One of the objects that is being merged. Refers to the version that is currently in use. (e.g. a displayed database record) update: The second object that is being merged. Refers to an update that needs to be integrated with the in-use version. default_dict_merge_op (:class:`json_merger.config.DictMergerOps` class attribute): Default strategy for merging regular non list JSON values (strings, numbers, other objects). default_list_merge_op (:class:`json_merger.config.UnifierOps` class attribute): Default strategy for merging two lists of entities. dict_merge_ops: Defines custom strategies for merging dict of entities. Dict formatted as: * keys -- a config string * values -- a class attribute of :class:`json_merger.config.DictMergerOps` list_merge_ops: Defines custom strategies for merging lists of entities. Dict formatted as: * keys -- a config string * values -- a class attribute of :class:`json_merger.config.UnifierOps` comparators: Defines classes used for rendering entities in list fields as equal. Dict formatted as: * keys -- a config string * values -- a class that extends :class:`json_merger.comparator.BaseComparator` data_lists: List of config strings defining the lists that are not treated as lists of entities. Note: A configuration string represents the path towards a list field in the object sepparated with dots. Example: Configuration strings can be: For ``{'lst': [{'id': 0, 'tags': ['foo', 'bar']}]}``: * the config string for the top level list is ``'lst'`` * the config string for the tags lists is ``'lst.tags'`` """ self.comparators = comparators or {} self.data_lists = set(data_lists or []) self.list_dict_ops = list_dict_ops or {} self.list_merge_ops = list_merge_ops or {} self.default_dict_merge_op = default_dict_merge_op self.default_list_merge_op = default_list_merge_op self.root = copy.deepcopy(root) self.head = copy.deepcopy(head) self.update = copy.deepcopy(update) self.head_stats = {} self.update_stats = {} self.conflicts = [] self.merged_root = None self.aligned_root = copy.deepcopy(root) self.aligned_head = copy.deepcopy(head) self.aligned_update = copy.deepcopy(update) def merge(self): """Populates result members. Performs the merge algorithm using the specified config and fills in the members that provide stats about the merging procedure. Attributes: merged_root: The result of the merge. aligned_root, aligned_head, aligned_update: Copies of root, head and update in which all matched list entities have the same list index for easier diff viewing. head_stats, update_stats: Stats for each list field present in the head or update objects. Instance of :class:`json_merger.stats.ListMatchStats` conflicts: List of :class:`json_merger.conflict.Conflict` instances that occured during the merge. Raises: :class:`json_merger.errors.MergeError` : If conflicts occur during the call. Example: >>> from json_merger import Merger >>> # We compare people by their name >>> from json_merger.comparator import PrimaryKeyComparator >>> from json_merger.config import DictMergerOps, UnifierOps >>> from json_merger.errors import MergeError >>> # Use this only for doctest :) >>> from pprint import pprint as pp >>> >>> root = {'people': [{'name': 'Jimmy', 'age': 30}]} >>> head = {'people': [{'name': 'Jimmy', 'age': 31}, ... {'name': 'George'}]} >>> update = {'people': [{'name': 'John'}, ... {'name': 'Jimmy', 'age': 32}]} >>> >>> class NameComparator(PrimaryKeyComparator): ... # Two objects are the same entitity if they have the ... # same name. ... primary_key_fields = ['name'] >>> m = Merger(root, head, update, ... DictMergerOps.FALLBACK_KEEP_HEAD, ... UnifierOps.KEEP_UPDATE_AND_HEAD_ENTITIES_HEAD_FIRST, ... comparators = {'people': NameComparator}) >>> # We do a merge >>> try: ... m.merge() ... except MergeError as e: ... # Conflicts are the same thing as the exception content. ... assert e.content == m.conflicts >>> # This is how the lists are aligned: >>> pp(m.aligned_root['people'], width=60) ['#$PLACEHOLDER$#', {'age': 30, 'name': 'Jimmy'}, '#$PLACEHOLDER$#'] >>> pp(m.aligned_head['people'], width=60) ['#$PLACEHOLDER$#', {'age': 31, 'name': 'Jimmy'}, {'name': 'George'}] >>> pp(m.aligned_update['people'], width=60) [{'name': 'John'}, {'age': 32, 'name': 'Jimmy'}, '#$PLACEHOLDER$#'] >>> # This is the end result of the merge: >>> pp(m.merged_root, width=60) {'people': [{'name': 'John'}, {'age': 31, 'name': 'Jimmy'}, {'name': 'George'}]} >>> # With some conflicts: >>> pp(m.conflicts, width=60) [('SET_FIELD', ('people', 1, 'age'), 32)] >>> # And some stats: >>> pp(m.head_stats[('people',)].in_result) [{'age': 31, 'name': 'Jimmy'}, {'name': 'George'}] >>> pp(m.update_stats[('people',)].not_in_result) [] Note: Even if conflicts occur, merged_root, aligned_root, aligned_head and aligned_update are always populated by following the startegies set for the merger instance. """ self.merged_root = self._recursive_merge(self.root, self.head, self.update) if self.conflicts: raise MergeError('Conflicts Occurred in Merge Process', self.conflicts) def _recursive_merge(self, root, head, update, key_path=()): dotted_key_path = get_dotted_key_path(key_path, filter_int_keys=True) if (isinstance(head, list) and isinstance(update, list) and dotted_key_path not in self.data_lists): # In this case we are merging two lists of objects. lists_to_unify = [()] if not isinstance(root, list): root = [] else: # Otherwise we merge everything but the lists using DictMergerOps. m = self._merge_objects(root, head, update, key_path) root = m.merged_root lists_to_unify = m.skipped_lists for list_field in lists_to_unify: absolute_key_path = key_path + list_field root_l = get_obj_at_key_path(root, list_field, []) head_l = get_obj_at_key_path(head, list_field, []) update_l = get_obj_at_key_path(update, list_field, []) unifier = self._unify_lists(root_l, head_l, update_l, absolute_key_path) new_list = [] for idx, objects in enumerate(unifier.unified): root_obj, head_obj, update_obj = objects new_obj = self._recursive_merge(root_obj, head_obj, update_obj, absolute_key_path + (idx, )) new_list.append(new_obj) root = set_obj_at_key_path(root, list_field, new_list) self._build_aligned_lists_and_stats(unifier, absolute_key_path) return root def _merge_objects(self, root, head, update, key_path): data_lists = get_conf_set_for_key_path(self.data_lists, key_path) object_merger = SkipListsMerger(root, head, update, self.default_dict_merge_op, data_lists, self.list_dict_ops, key_path) try: object_merger.merge() except MergeError as e: self.conflicts.extend(c.with_prefix(key_path) for c in e.content) return object_merger def _unify_lists(self, root, head, update, key_path): dotted_key_path = get_dotted_key_path(key_path, True) operation = self.list_merge_ops.get(dotted_key_path, self.default_list_merge_op) comparator_cls = self.comparators.get(dotted_key_path, DefaultComparator) list_unifier = ListUnifier(root, head, update, operation, comparator_cls) try: list_unifier.unify() except MergeError as e: self.conflicts.extend(c.with_prefix(key_path) for c in e.content) return list_unifier def _build_aligned_lists_and_stats(self, list_unifier, key_path): root_list = [] head_list = [] update_list = [] for root_obj, head_obj, update_obj in list_unifier.unified: # Cast NOTHING objects to a placeholder so we reserialize back to # JSON if needed. root_list.append(root_obj or PLACEHOLDER_STR) head_list.append(head_obj or PLACEHOLDER_STR) update_list.append(update_obj or PLACEHOLDER_STR) # Try to put back the list if the key path existed in the first place. self.aligned_root = set_obj_at_key_path(self.aligned_root, key_path, root_list, False) self.aligned_head = set_obj_at_key_path(self.aligned_head, key_path, head_list, False) self.aligned_update = set_obj_at_key_path(self.aligned_update, key_path, update_list, False) # Also copy over the stats. self.head_stats[key_path] = list_unifier.head_stats self.update_stats[key_path] = list_unifier.update_stats