X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=dict_utils.py;h=6f0f572e988cd5fdfea2e52c1fe1da8e50a80188;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=7b0edb50a87a5854f20326408cfc0a03d22408dc;hpb=d08bad64a6884f25d28a2c38c6cd1c87b4335188;p=python_utils.git diff --git a/dict_utils.py b/dict_utils.py index 7b0edb5..6f0f572 100644 --- a/dict_utils.py +++ b/dict_utils.py @@ -1,5 +1,9 @@ #!/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 @@ -9,7 +13,7 @@ def init_or_inc( 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 @@ -38,14 +42,16 @@ def shard(d: Dict[Any, Any], size: int) -> Iterator[Dict[Any, Any]]: """ Shards a dict into N subdicts which, together, contain all keys/values from the original unsharded dict. + """ 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): from list_utils import flatten + return flatten([new_value, old_value]) @@ -53,11 +59,11 @@ def coalesce_by_creating_set(key, new_value, old_value): 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): return new_value -def coalesce_first_write_wins(key, new_value, old_value): +def coalesce_first_write_wins(_, discarded_new_value, old_value): return old_value @@ -66,9 +72,9 @@ def raise_on_duplicated_keys(key, new_value, old_value): 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 @@ -221,7 +227,7 @@ def dict_to_key_value_lists(d: Dict[Any, Any]) -> Tuple[List[Any], List[Any]]: ['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) @@ -230,4 +236,5 @@ def dict_to_key_value_lists(d: Dict[Any, Any]) -> Tuple[List[Any], List[Any]]: if __name__ == '__main__': import doctest + doctest.testmod()