from functools import reduce from itertools import groupby from operator import iconcat from typing import TypeVar, List, Dict, Optional, Any, Iterable from typing import Callable, Generator T = TypeVar("T") NewKey = TypeVar("NewKey") Key = TypeVar("Key") Value = TypeVar("Value") def any_present(items: Iterable[T], check_items: Iterable[T]): if items is None: return False return any(item in check_items for item in items) def first_or_none(items: List[T], predicate) -> T: return next((value for value in items if predicate(value)), None) def unique(items: List[T]) -> List[T]: return list(set(items)) def flattern(items: List[List[T]]) -> List[T]: return reduce(iconcat, items, []) def first_or_default(items: List[T], predicate: Callable[[T], bool], default: T) -> T: return next((value for value in items if predicate(value)), default) def range_chunks(itr, n: int) -> Generator[List[T], Any, None]: """Yield successive n-sized chunks from lst.""" for i in range(0, len(itr), n): yield itr[i: i + n] def chunks(lst: List[T], n: int) -> List[T]: return list(range_chunks(lst, n)) def map_dict_keys(dict_to_map: Dict[Key, Value], map_func: Callable[[Key], NewKey]) -> Dict[NewKey, Value]: clean_dict = {} for key, value in dict_to_map.items(): clean_dict[map_func(key)] = value return clean_dict def group_by(items: List[T], key_extractor) -> Dict[str, List[T]]: sorted_results = sorted(items, key=key_extractor) grouped_results = groupby(sorted_results, key=key_extractor) groups: Dict[str, List[T]] = {} for k, g in grouped_results: groups[k] = list(g) return groups def exists(items: List[T], predicate: Callable[[T], bool]) -> bool: return first_or_none(items, predicate) is not None def has_changes(original: List, updated: List) -> bool: original_set = set(original) updated_set = set(updated) return len(original_set.symmetric_difference(updated_set)) > 0 def index_of(items: List[T], comparator: Callable[[T], bool]) -> int: for index, item in enumerate(items): if comparator(item): return index raise IndexError("Item not found in the list") def get(items: List[T], index: int) -> Optional[T]: return items[index] if len(items) > index else None def safe_sum(items: List[T], default: Any = None) -> Optional[T]: items = filter(None, items) return sum(items) if items else default def get_nested_attr(target: object, path: str, *args) -> Any: """ get_nested_attr(target, path[, default]) -> value Use to safely get nested objects data. Raises AttributeError exception, if attribute is not found and default is not defined """ return reduce(lambda obj, attr: getattr(obj, attr, *args), path.split("."), target) def agg_by(items: List[T], extractor: Callable[[T], Key]) -> Dict[Key, List[T]]: agg_data = dict() for item in items: key = extractor(item) if key not in agg_data: agg_data[key] = [item] else: agg_data[key] += [item] return agg_data def diff(lst: List, comparator: List) -> List: """ Function for compare to list and returning difference between. Args: lst: comparator: Returns: List of difference. Empty list if lists the same """ return list(set(lst).symmetric_difference(set(comparator)))