from collections import namedtuple from typing import Any, Callable, Dict, Iterable, List, Union Processor = namedtuple("Processor", ("func", "args", "kwargs")) def processor(func: Callable, *args, **kwargs) -> Processor: """Create processor instance""" if not func: raise ValueError("func is required.") return Processor(func=func, args=(args or tuple()), kwargs=(kwargs or {})) def apply_processors(processors: Iterable[Union[Callable, Processor]], item: Dict[str, Any], item_id: str = None): for p in processors: args = (item,) if item_id is None else (item, item_id) if isinstance(p, Processor): p.func(*args, *p.args, **p.kwargs) else: p(*args) def process_by_item( result: Union[Dict[str, Dict[str, Any]], List[Dict[str, Any]]], processors: Iterable[Union[Callable, Processor]], id_getter: Callable[[Any], Any] = None, ): """Wrapper to organize sequential processors on every item of iterated collection in one loop.""" if isinstance(result, dict): for item_id, item in result.items(): apply_processors(processors, item, item_id) return for item in result: item_id = id_getter(item) if id_getter else None apply_processors(processors, item, item_id) def apply_by_item( result: Union[Dict[str, Dict[str, Any]], List[Dict[str, Any]]], processors: Iterable[Union[Callable, Processor]] = None, processors_by_id: Iterable[Union[Callable, Processor]] = None, id_getter: Callable[[Any], Any] = None, ): if processors_by_id and not id_getter: raise ValueError("id_getter is required for processors_with_id") if not processors and not processors_by_id: return result if isinstance(result, dict): for item_id, item in result.items(): apply_processors(processors + processors_by_id, item, item_id) return for item in result: apply_processors(processors, item) apply_processors(processors_by_id, item, id_getter(item))