3 # © Copyright 2021-2022, Scott Gasch
5 """Helper functions for dealing with dictionaries."""
7 from itertools import islice
8 from typing import Any, Callable, Dict, Iterator, List, Tuple
16 inc_function: Callable[..., Any] = lambda x: x + 1,
19 Initialize a dict value (if it doesn't exist) or increments it (using the
20 inc_function, which is customizable) if it already does exist. Returns
21 True if the key already existed or False otherwise.
24 >>> init_or_inc(d, "test")
26 >>> init_or_inc(d, "test")
28 >>> init_or_inc(d, 'ing')
35 d[key] = inc_function(d[key])
41 def shard(d: Dict[Any, Any], size: int) -> Iterator[Dict[Any, Any]]:
43 Shards a dict into N subdicts which, together, contain all keys/values
44 from the original unsharded dict.
48 for x in range(0, len(d), size):
49 yield dict(islice(items, x, x + size))
52 def coalesce_by_creating_list(_, new_value, old_value):
53 from list_utils import flatten
55 return flatten([new_value, old_value])
58 def coalesce_by_creating_set(key, new_value, old_value):
59 return set(coalesce_by_creating_list(key, new_value, old_value))
62 def coalesce_last_write_wins(_, new_value, discarded_old_value):
66 def coalesce_first_write_wins(_, discarded_new_value, old_value):
70 def raise_on_duplicated_keys(key, new_value, old_value):
71 raise Exception(f'Key {key} is duplicated in more than one input dict.')
75 inputs: Iterator[Dict[Any, Any]],
77 aggregation_function: Callable[[Any, Any, Any], Any] = coalesce_by_creating_list,
79 """Merge N dicts into one dict containing the union of all keys /
80 values in the input dicts. When keys collide, apply the
81 aggregation_function which, by default, creates a list of values.
82 See also several other alternative functions for coalescing values
83 (coalesce_by_creating_set, coalesce_first_write_wins,
84 coalesce_last_write_wins, raise_on_duplicated_keys) or provide a
85 custom helper function.
87 >>> a = {'a': 1, 'b': 2}
88 >>> b = {'b': 1, 'c': 2, 'd': 3}
89 >>> c = {'c': 1, 'd': 2}
90 >>> coalesce([a, b, c])
91 {'a': 1, 'b': [1, 2], 'c': [1, 2], 'd': [2, 3]}
93 >>> coalesce([a, b, c], aggregation_function=coalesce_last_write_wins)
94 {'a': 1, 'b': 1, 'c': 1, 'd': 2}
96 >>> coalesce([a, b, c], aggregation_function=raise_on_duplicated_keys)
97 Traceback (most recent call last):
99 Exception: Key b is duplicated in more than one input dict.
102 out: Dict[Any, Any] = {}
106 value = aggregation_function(key, d[key], out[key])
113 def item_with_max_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
114 """Returns the key and value with the max value in a dict.
116 >>> d = {'a': 1, 'b': 2, 'c': 3}
117 >>> item_with_max_value(d)
119 >>> item_with_max_value({})
120 Traceback (most recent call last):
122 ValueError: max() arg is an empty sequence
125 return max(d.items(), key=lambda _: _[1])
128 def item_with_min_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
129 """Returns the key and value with the min value in a dict.
131 >>> d = {'a': 1, 'b': 2, 'c': 3}
132 >>> item_with_min_value(d)
136 return min(d.items(), key=lambda _: _[1])
139 def key_with_max_value(d: Dict[Any, Any]) -> Any:
140 """Returns the key with the max value in the dict.
142 >>> d = {'a': 1, 'b': 2, 'c': 3}
143 >>> key_with_max_value(d)
147 return item_with_max_value(d)[0]
150 def key_with_min_value(d: Dict[Any, Any]) -> Any:
151 """Returns the key with the min value in the dict.
153 >>> d = {'a': 1, 'b': 2, 'c': 3}
154 >>> key_with_min_value(d)
158 return item_with_min_value(d)[0]
161 def max_value(d: Dict[Any, Any]) -> Any:
162 """Returns the maximum value in the dict.
164 >>> d = {'a': 1, 'b': 2, 'c': 3}
169 return item_with_max_value(d)[1]
172 def min_value(d: Dict[Any, Any]) -> Any:
173 """Returns the minimum value in the dict.
175 >>> d = {'a': 1, 'b': 2, 'c': 3}
180 return item_with_min_value(d)[1]
183 def max_key(d: Dict[Any, Any]) -> Any:
184 """Returns the maximum key in dict (ignoring values totally)
186 >>> d = {'a': 3, 'b': 2, 'c': 1}
194 def min_key(d: Dict[Any, Any]) -> Any:
195 """Returns the minimum key in dict (ignoring values totally)
197 >>> d = {'a': 3, 'b': 2, 'c': 1}
205 def parallel_lists_to_dict(keys: List[Any], values: List[Any]) -> Dict[Any, Any]:
206 """Given two parallel lists (keys and values), create and return
209 >>> k = ['name', 'phone', 'address', 'zip']
210 >>> v = ['scott', '555-1212', '123 main st.', '12345']
211 >>> parallel_lists_to_dict(k, v)
212 {'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
215 if len(keys) != len(values):
216 raise Exception("Parallel keys and values lists must have the same length")
217 return dict(zip(keys, values))
220 def dict_to_key_value_lists(d: Dict[Any, Any]) -> Tuple[List[Any], List[Any]]:
222 >>> d = {'name': 'scott', 'phone': '555-1212', 'address': '123 main st.', 'zip': '12345'}
223 >>> (k, v) = dict_to_key_value_lists(d)
225 ['name', 'phone', 'address', 'zip']
227 ['scott', '555-1212', '123 main st.', '12345']
230 r: Tuple[List[Any], List[Any]] = ([], [])
231 for (k, v) in d.items():
237 if __name__ == '__main__':