|
2 | 2 | from collections import abc |
3 | 3 | from typing import Any, Dict, List, Mapping, Optional, Type, TypeVar |
4 | 4 |
|
| 5 | +from deepmerge import Merger |
| 6 | + |
5 | 7 | from config.errors import ConfigurationOverrideError |
6 | 8 |
|
7 | 9 | T = TypeVar("T") |
8 | 10 |
|
| 11 | +merger = Merger( |
| 12 | + type_strategies=[ |
| 13 | + (list, ["append"]), |
| 14 | + (dict, ["merge"]), |
| 15 | + (set, ["union"]), |
| 16 | + ], |
| 17 | + fallback_strategies=["override"], |
| 18 | + type_conflict_strategies=["override"], |
| 19 | +) |
| 20 | + |
9 | 21 |
|
10 | | -def apply_key_value(obj, key, value): |
| 22 | +def apply_key_value(obj: Mapping[str, Any], key: str, value: Any) -> Mapping[str, Any]: |
11 | 23 | key = key.strip("_:.") # remove special characters from both ends |
12 | 24 | for token in (":", "__", "."): |
13 | 25 | if token in key: |
@@ -52,27 +64,33 @@ def apply_key_value(obj, key, value): |
52 | 64 | ) |
53 | 65 |
|
54 | 66 | try: |
55 | | - sub_property[index] = value |
| 67 | + sub_property[index] = merger.merge(sub_property[index], value) |
56 | 68 | except IndexError: |
57 | 69 | raise ConfigurationOverrideError( |
58 | 70 | f"Invalid override for mutable sequence {key}; " |
59 | 71 | f"assignment index out of range" |
60 | 72 | ) |
61 | 73 | else: |
62 | 74 | try: |
63 | | - sub_property[last_part] = value |
| 75 | + if isinstance(sub_property, abc.Mapping): |
| 76 | + sub_property[last_part] = merger.merge( |
| 77 | + sub_property.get(last_part), |
| 78 | + value, |
| 79 | + ) |
| 80 | + else: |
| 81 | + sub_property[last_part] = value |
64 | 82 | except TypeError as type_error: |
65 | 83 | raise ConfigurationOverrideError( |
66 | 84 | f"Invalid assignment {key} -> {value}; {str(type_error)}" |
67 | 85 | ) |
68 | 86 |
|
69 | 87 | return obj |
70 | 88 |
|
71 | | - obj[key] = value |
| 89 | + obj[key] = merger.merge(obj.get(key), value) |
72 | 90 | return obj |
73 | 91 |
|
74 | 92 |
|
75 | | -def merge_values(destination, source): |
| 93 | +def merge_values(destination: Mapping[str, Any], source: Mapping[str, Any]) -> None: |
76 | 94 | for key, value in source.items(): |
77 | 95 | apply_key_value(destination, key, value) |
78 | 96 |
|
|
0 commit comments