"""Utils.""" import json from logger import logger import os from typing import Any, Dict, Tuple def execute_chunks(chunk_size: int): """Split list on chunks and execute function for each. Args: chunk_size (int): Split on chunks of this size. """ def decorator(f): def wrapper(items: list, *args, **kwargs): result = None for i in range(0, len(items), chunk_size): logger.info('Process chunk {}-{} of {}'.format( i, i + chunk_size - 1, len(items))) items_chunk = items[i:i + chunk_size] result_chunk = f(items_chunk, *args, **kwargs) if result_chunk: if not result: result = result_chunk elif isinstance(result_chunk, list): result.extend(result_chunk) elif isinstance(result_chunk, dict): result.update(result_chunk) else: raise NotImplementedError() return result return wrapper return decorator def check_dir(path: str): """Check dir exist, create if not. Args: path (str): Dir path. """ if not os.path.isdir(path): os.mkdir(path) def check_files_exist(*args: Tuple[str]) -> bool: """Check all files exist. Args: args (tuple): List of files. Return: bool: All exist or not. """ result = True for filename in args: result = result and os.path.isfile(filename) return result def read_json_files(*args: Tuple[str]) -> tuple: """Read set of json files. Args: args (tuple): List of files. Returns: tuple: Files content. """ results = [] for filename in args: with open(filename, 'r') as infile: results.append(json.load(infile)) return tuple(results) def write_json_files(**kwargs: Dict[str, Any]): """Write set of json data to files. Args: kwargs (dict): File name and its content. """ for filename, data in kwargs.items(): with open(filename, 'w') as outfile: json.dump(data, outfile)