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