#!/usr/bin/env python3
+# © Copyright 2021-2022, Scott Gasch
+
+"""Helper functions for dealing with dictionaries."""
+
from itertools import islice
from typing import Any, Callable, Dict, Iterator, List, Tuple
key: Any,
*,
init_value: Any = 1,
- inc_function: Callable[..., Any] = lambda x: x + 1
+ inc_function: Callable[..., Any] = lambda x: x + 1,
) -> bool:
"""
Initialize a dict value (if it doesn't exist) or increments it (using the
"""
items = d.items()
for x in range(0, len(d), size):
- yield {key: value for (key, value) in islice(items, x, x + size)}
+ yield dict(islice(items, x, x + size))
-def coalesce_by_creating_list(key, new_value, old_value):
+def coalesce_by_creating_list(_, new_value, old_value):
+ """Helper for use with :meth:`coalesce` that creates a list on
+ collision."""
from list_utils import flatten
+
return flatten([new_value, old_value])
def coalesce_by_creating_set(key, new_value, old_value):
+ """Helper for use with :meth:`coalesce` that creates a set on
+ collision."""
return set(coalesce_by_creating_list(key, new_value, old_value))
-def coalesce_last_write_wins(key, new_value, old_value):
+def coalesce_last_write_wins(_, new_value, discarded_old_value):
+ """Helper for use with :meth:`coalsce` that klobbers the old
+ with the new one on collision."""
return new_value
-def coalesce_first_write_wins(key, new_value, old_value):
+def coalesce_first_write_wins(_, discarded_new_value, old_value):
+ """Helper for use with :meth:`coalsce` that preserves the old
+ value and discards the new one on collision."""
return old_value
def raise_on_duplicated_keys(key, new_value, old_value):
+ """Helper for use with :meth:`coalesce` that raises an exception
+ when a collision is detected.
+ """
raise Exception(f'Key {key} is duplicated in more than one input dict.')
def coalesce(
- inputs: Iterator[Dict[Any, Any]],
- *,
- aggregation_function: Callable[[Any, Any], Any] = coalesce_by_creating_list
+ inputs: Iterator[Dict[Any, Any]],
+ *,
+ aggregation_function: Callable[[Any, Any, Any], Any] = coalesce_by_creating_list,
) -> Dict[Any, Any]:
"""Merge N dicts into one dict containing the union of all keys /
values in the input dicts. When keys collide, apply the
aggregation_function which, by default, creates a list of values.
- See also several other alternative functions for coalescing values
- (coalesce_by_creating_set, coalesce_first_write_wins,
- coalesce_last_write_wins, raise_on_duplicated_keys) or provide a
- custom helper function.
+ See also several other alternative functions for coalescing values:
+
+ * :meth:`coalesce_by_creating_set`
+ * :meth:`coalesce_first_write_wins`
+ * :meth:`coalesce_last_write_wins`
+ * :meth:`raise_on_duplicated_keys`
+ * or provive your own collision resolution code.
>>> a = {'a': 1, 'b': 2}
>>> b = {'b': 1, 'c': 2, 'd': 3}
def item_with_max_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
- """Returns the key and value with the max value in a dict.
+ """Returns the key and value of the item with the max value in a dict.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> item_with_max_value(d)
def item_with_min_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
- """Returns the key and value with the min value in a dict.
+ """Returns the key and value of the item with the min value in a dict.
>>> d = {'a': 1, 'b': 2, 'c': 3}
>>> item_with_min_value(d)
['scott', '555-1212', '123 main st.', '12345']
"""
- r = ([], [])
+ r: Tuple[List[Any], List[Any]] = ([], [])
for (k, v) in d.items():
r[0].append(k)
r[1].append(v)
if __name__ == '__main__':
import doctest
+
doctest.testmod()