dict_utils.py

   1 #!/usr/bin/env python3
   2
   3 from itertools import islice
   4 from typing import Any, Callable, Dict, Iterator, Tuple
   5
   6
   7 def init_or_inc(
   8     d: Dict[Any, Any],
   9     key: Any,
  10     *,
  11     init_value: Any = 1,
  12     inc_function: Callable[..., Any] = lambda x: x + 1
  13 ) -> bool:
  14     if key in d.keys():
  15         d[key] = inc_function(d[key])
  16         return True
  17     d[key] = init_value
  18     return False
  19
  20
  21 def shard(d: Dict[Any, Any], size: int) -> Iterator[Dict[Any, Any]]:
  22     items = d.items()
  23     for x in range(0, len(d), size):
  24         yield {key: value for (key, value) in islice(items, x, x + size)}
  25
  26
  27 def coalesce_by_creating_list(key, v1, v2):
  28     from list_utils import flatten
  29     return flatten([v1, v2])
  30
  31
  32 def coalesce_by_creating_set(key, v1, v2):
  33     return set(coalesce_by_creating_list(key, v1, v2))
  34
  35
  36 def raise_on_duplicated_keys(key, v1, v2):
  37     raise Exception(f'Key {key} is duplicated in more than one input dict.')
  38
  39
  40 def coalesce(
  41         inputs: Iterator[Dict[Any, Any]],
  42         *,
  43         aggregation_function: Callable[[Any, Any], Any] = coalesce_by_creating_list
  44 ) -> Dict[Any, Any]:
  45     out: Dict[Any, Any] = {}
  46     for d in inputs:
  47         for key in d:
  48             if key in out:
  49                 value = aggregation_function(d[key], out[key])
  50             else:
  51                 value = d[key]
  52             out[key] = value
  53     return out
  54
  55
  56 def item_with_max_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
  57     return max(d.items(), key=lambda _: _[1])
  58
  59
  60 def item_with_min_value(d: Dict[Any, Any]) -> Tuple[Any, Any]:
  61     return min(d.items(), key=lambda _: _[1])
  62
  63
  64 def key_with_max_value(d: Dict[Any, Any]) -> Any:
  65     return item_with_max_value(d)[0]
  66
  67
  68 def key_with_min_value(d: Dict[Any, Any]) -> Any:
  69     return item_with_min_value(d)[0]
  70
  71
  72 def max_value(d: Dict[Any, Any]) -> Any:
  73     return item_with_max_value(d)[1]
  74
  75
  76 def min_value(d: Dict[Any, Any]) -> Any:
  77     return item_with_min_value(d)[1]
  78
  79
  80 def max_key(d: Dict[Any, Any]) -> Any:
  81     return max(d.keys())
  82
  83
  84 def min_key(d: Dict[Any, Any]) -> Any:
  85     return min(d.keys())