from typing import Any, Dict, List, Mapping, Optional, Sequence from .collections_diff_types import RecordChange, RecordDiff # Largest integer JavaScript can hold exactly; beyond it the three languages disagree. SAFE_INTEGER = 9007199254740991 def _key_of(value: Any, key: str) -> Optional[str]: """A record's key as text, rendered the way collections.group-by-key renders a group name so the two agree on what "the same key" means. None when absent. """ if value is None: return None if isinstance(value, str): return value # bool before int: in Python True is an int. if isinstance(value, bool): return "true" if value else "false" if isinstance(value, int): if abs(value) > SAFE_INTEGER: raise ValueError('cannot diff by the out-of-range number %d at "%s"' % (value, key)) return str(value) if isinstance(value, float): raise TypeError('cannot diff by the fractional number %r at "%s"' % (value, key)) raise TypeError('cannot diff by the list or map at "%s"' % (key,)) def _same(a: Any, b: Any) -> bool: """Deep JSON equality, spelled out so all three languages agree: numbers by value (1 equals 1.0), no coercion between types (True is not 1, "1" is not 1), lists in order, and a missing map field equal to a None one. """ if a is None or b is None: return a is None and b is None # Python's == says True == 1; JSON does not. if isinstance(a, bool) or isinstance(b, bool): return isinstance(a, bool) and isinstance(b, bool) and a == b if isinstance(a, (int, float)): return isinstance(b, (int, float)) and a == b if isinstance(a, str): return isinstance(b, str) and a == b if isinstance(a, (list, tuple)): return ( isinstance(b, (list, tuple)) and len(a) == len(b) and all(_same(x, y) for x, y in zip(a, b)) ) if isinstance(a, dict) and isinstance(b, dict): return all(_same(a.get(k), b.get(k)) for k in set(a) | set(b)) return False def _index(records: Sequence[Mapping[str, Any]], key: str, side: str) -> Dict[str, int]: """Key every record of one list, refusing missing and duplicate keys.""" by_key: Dict[str, int] = {} for i, record in enumerate(records): k = _key_of(record.get(key) if isinstance(record, dict) else None, key) if k is None: raise TypeError('record %d in %s has no value at "%s"' % (i, side, key)) # Picking one of two duplicates would report changes that never happened. if k in by_key: raise ValueError('duplicate key "%s" in %s' % (k, side)) by_key[k] = i return by_key def diff_by_key( before: Sequence[Mapping[str, Any]], after: Sequence[Mapping[str, Any]], key: str, ) -> RecordDiff[Mapping[str, Any]]: """Compare two lists of records matched by ``key``: what was added, what was removed, and what changed and in which fields. """ for side in (before, after): if isinstance(side, (str, bytes)) or not isinstance(side, (list, tuple)): raise TypeError("diff_by_key needs two lists of records") if not isinstance(key, str) or key == "": raise TypeError("diff_by_key needs a non-empty key name") before_keys = _index(before, key, "before") after_keys = _index(after, key, "after") added: List[Mapping[str, Any]] = [] changed: List[RecordChange[Mapping[str, Any]]] = [] unchanged = 0 for k, i in after_keys.items(): nxt = after[i] j = before_keys.get(k) if j is None: added.append(nxt) continue prev = before[j] names = set(prev) | set(nxt) # sorted() on str is code point order, the order all three pin. fields = sorted(name for name in names if not _same(prev.get(name), nxt.get(name))) if fields: changed.append(RecordChange(key=k, before=prev, after=nxt, fields=fields)) else: unchanged += 1 removed = [before[j] for k, j in before_keys.items() if k not in after_keys] return RecordDiff(added=added, removed=removed, changed=changed, unchanged=unchanged)