Coverage for jsonschema_diff/core/tools/combine.py: 100%

1from collections import OrderedDict 

2from typing import Any, Dict, List, Tuple, TypeAlias 

3 

4COMBINE_RULES_TYPE: TypeAlias = List[List[str]] 

5"""Rule format: each inner list declares a logical group of keys.""" 

6 

7 

8class LogicCombinerHandler: 

9 """Group items by user-defined rules and merge their inner fields.""" 

10 

11 # ------------------------------------------------------------------ # 

12 # Internal helpers 

13 # ------------------------------------------------------------------ # 

14 

15 @staticmethod 

16 def _require_inner_fields(inner_key_field: str | None, inner_value_field: str | None) -> None: 

17 if not inner_key_field or not inner_value_field: 

18 raise ValueError("inner_key_field и inner_value_field должны быть заданы.") 

19 

20 @staticmethod 

21 def _extract( 

22 item: Any, key_name: str, inner_key_field: str, inner_value_field: str 

23 ) -> Tuple[Any, Any]: 

24 """ 

25 Return ``(inner_key, inner_value)`` taken from ``item`` (a dict). 

26 

27 Parameters 

28 ---------- 

29 item : Any 

30 Mapping that must contain both inner fields. 

31 key_name : str 

32 Name of *item* inside the outer ``subset`` (used for messages). 

33 inner_key_field, inner_value_field : str 

34 Mandatory keys to pull out. 

35 

36 Raises 

37 ------ 

38 TypeError 

39 If *item* is not a dict or lacks required fields. 

40 """ 

41 if not isinstance(item, dict): 

42 raise TypeError(f"Expected dict for '{key_name}', got {type(item).__name__}") 

43 if inner_key_field not in item or inner_value_field not in item: 

44 raise TypeError( 

45 f"Item '{key_name}' must contain '{inner_key_field}' and '{inner_value_field}'" 

46 ) 

47 return item[inner_key_field], item[inner_value_field] 

48 

49 # ------------------------------------------------------------------ # 

50 # Public API 

51 # ------------------------------------------------------------------ # 

52 

53 @staticmethod 

54 def combine( 

55 subset: Dict[str, Any], 

56 rules: List[List[str]], 

57 inner_key_field: str = "comparator", 

58 inner_value_field: str = "to_compare", 

59 ) -> Dict[Tuple[str, ...], Dict[str, Any]]: 

60 """ 

61 Build an ``OrderedDict`` that groups *subset* items per *rules*. 

62 

63 Returns 

64 ------- 

65 dict 

66 ``(k1, k2, …) -> {inner_key_field: common_key, inner_value_field: [v1, v2, …]}`` 

67 

68 Note 

69 ---- 

70 * Keys not covered by *rules* stay as single-element groups. 

71 * Inner keys in the same group must match or ``ValueError`` is raised. 

72 """ 

73 LogicCombinerHandler._require_inner_fields(inner_key_field, inner_value_field) 

74 out: "OrderedDict[Tuple[str, ...], Dict[str, Any]]" = OrderedDict() 

75 seen_in_rules: set[str] = set() 

76 

77 # 1. groups coming from explicit rules 

78 for rule in rules: 

79 present = [k for k in rule if k in subset] 

80 if not present: 

81 continue 

82 

83 fields, vals = [], [] 

84 for k in present: 

85 f, v = LogicCombinerHandler._extract( 

86 subset[k], k, inner_key_field, inner_value_field 

87 ) 

88 fields.append(f) 

89 vals.append(v) 

90 

91 base_field = fields[0] 

92 if any(f != base_field for f in fields[1:]): 

93 raise ValueError(f"Mismatched '{inner_key_field}' inside group {tuple(present)}") 

94 

95 out[tuple(present)] = {inner_key_field: base_field, inner_value_field: vals} 

96 seen_in_rules.update(present) 

97 

98 # 2. leftover singletons, keep original order 

99 for k, item in subset.items(): 

100 if k in seen_in_rules: 

101 continue 

102 f, v = LogicCombinerHandler._extract(item, k, inner_key_field, inner_value_field) 

103 out[(k,)] = {inner_key_field: f, inner_value_field: [v]} 

104 

105 return out