"""Combine logic to handle nested dictionaries.""" import glom from collections import namedtuple from copy import deepcopy from typing import Any, Callable, Dict, Iterable, List, Sequence def add_by_indexes(one: Iterable[int], another: Iterable[int]) -> tuple: """Add two iterable by indexes.""" if not one or not another: return tuple(one or another or []) return tuple(map(sum, zip(one, another))) def add_by_keys( result: Dict[str, Any], another: Dict[str, Any], excluded_keys: Iterable = None, included_keys: Iterable = None, none_as: Any = None, ): """Sums up two dictionaries data by all keys in first one. Can work with nested dictionaries, nested flat lists and numbers. """ def _wrap() -> Callable: def _not_wrapped(_value): return _value def _wrapped(_value): return _value if _value is not None else none_as return _not_wrapped if none_as is None else _wrapped def _add_nullable(_a, _b): if _a == none_as: if _b == none_as: return none_as return _b if _b == none_as: return _a return _a + _b def _add_by_keys(_result, _another, excluded_keys_set, included_keys_set): wrapped = _wrap() for k, v in _another.items(): if k in excluded_keys_set or included_keys_set and k not in included_keys_set: continue combined = _result.get(k) v = wrapped(v) if not combined: _result[k] = v continue if isinstance(v, dict): _add_by_keys(combined, v, excluded_keys_set, included_keys_set) elif isinstance(v, list): _result[k] = add_by_indexes(combined, v) else: _result[k] = _add_nullable(wrapped(_result[k]), v) excluded_keys = set(excluded_keys or []) included_keys = set(included_keys or []) _add_by_keys(result, another, excluded_keys, included_keys) def get_split_values(target: dict, result: list, depth: int): """Recursive function to transform target dictionary to list of nested dictionaries. Does not change the passed dictionary target. Places the result in 'result' list passed. Args: target: dictionary with nested dictionaries inside to get flat data from. result: list for place the result in, empty list should be passed on the top level. depth: level of nesting which should be flatten. Examples: before running: target = { "a1": { "b11": {"c111" 111}, "b12": {"c121" 121}, }, "a2": { "b21": {"c221" 221}, "b22": {"c222" 222}, } } result = [] depth = 2 after running get_flat_values(target, depth, result): result -> [{"c111" 111}, {"c121" 121}, {"c221" 221}, {"c222" 222}] """ if depth <= 1: result.extend(target.values()) return depth -= 1 for v in target.values(): get_split_values(v, result, depth) # Rule to specify mapping for combined values in "get_combined_data". Use combine_rule factory function for creation. _CombineRule = namedtuple( "_CombineRule", ("key_from", "key_to", "value", "inner_keys", "key_is_changed", "inner_is_flat"), defaults=("", None, None, None, False, False), ) def combine_rule(key_from: str, key_to: str = None, value: Any = None, inner_keys: Sequence = None) -> _CombineRule: """Create _CombineRule instance. There are 3 ways of usage: 1. combine_rule(key_from="a", key_to="b") - to change the key name in the result 2. combine_rule(key_from="a", value="some_combined_value") - to set value for key in combined result, "key_to" is optional here. 3. combine_rule(key_from="a", inner_keys=("sum1", "sum2")) - to set merged value in combined result, "key_to" is optional here, check examples in get_combined_data description. Args: key_from: key name of combined value in original data. key_to: optional key name in combined data for combined value. If keys in original and combined data are the same it is sufficient to define "key_from" only. value: optional known value to set by key in combined data. inner_keys: optional iterable of original data keys to be included in merged attribute of combined data, check an example. Be careful, original item should contain all inner keys and all values found by inner keys in the item should be hashable. """ _key_to = key_to or key_from key_is_changed = _key_to != key_from inner_keys = list(set(inner_keys or [])) inner_is_flat = len(inner_keys) == 1 if not key_is_changed and not value and not inner_keys: raise ValueError( f"Invalid combine rule: key_from={key_from}, key_to={key_to}, value={value}, inner_keys={inner_keys}." ) return _CombineRule( key_from=key_from, key_to=_key_to, value=value, inner_keys=inner_keys, key_is_changed=key_is_changed, inner_is_flat=inner_is_flat, ) def get_combined_data( items: Iterable, group_keys: Iterable, excluded_keys: Iterable = None, combine_rules: List[_CombineRule] = None ) -> List: """Sum up values from passed items for specific keys. Args: items: iterable of dictionaries to sum up data from. group_keys: iterable of keys to group by. combine_rules: rules of combining, defining how to rename|set|merge data for particular keys. excluded_keys: keys for which data should not be summed. Elements from group_keys and rule.key_from from combine_rules will be included in excluded_keys, there is no need to specify them. Examples: before running: items = [ { "id": "id1", "a": "VAL1", "b": { "c": 1, "d": 1 } "color": "red" "sum1": 0, "sum2": 5 }, { "id": "id2", "a": "VAL1", "b": { "c": 2, "d": 3 }, "color": "blue" "sum1": 1, "sum2": 1 } ] after running get_combined_data( items, group_keys=('a',), excluded_keys=('c',), combine_rules=( combine_rule(key_from="id", value="id1&2"), combine_rule(key_from="color", key_to="sum_by_color", inner_keys=("sum1", "sum2"))) ) ): result -> [{ "id": "id1&2", "a": "VAL1", "b": { "d": 4 }, "sum1": 1, "sum2": 6, "sum_by_color": { "red": { "sum1": 0, "sum2": 5 }, "blue: { "sum1": 1, "sum2": 1 } } }] """ def get_spec(_item: Dict[str, Any]) -> str: return ".".join([_item[key] for key in group_keys]) excluded_keys = {*group_keys, *(excluded_keys or []), *[r.key_from for r in (combine_rules or [])]} result = {} for item in deepcopy(items): for rule in combine_rules or []: value = item[rule.key_from] key_to = rule.key_to if rule.key_is_changed: del item[rule.key_from] if rule.inner_keys: merged_value = item[rule.inner_keys[0]] if rule.inner_is_flat else {k: item[k] for k in rule.inner_keys} key_to = f"{key_to}.{value}" else: merged_value = rule.value if rule.value else value glom.glom(item, glom.Assign(key_to, merged_value, missing=dict)) spec = get_spec(item) combined = glom.glom(result, spec, default=None) if combined: add_by_keys(combined, item, excluded_keys=excluded_keys) continue glom.glom(result, glom.Assign(spec, item, missing=dict)) flat_result = [] get_split_values(result, flat_result, depth=len(group_keys)) return flat_result