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