import time from functools import wraps from typing import Callable from apollo_notifications.config import LoopRunningConfig def time_logger(logger, app_name: str, **_kwargs) -> Callable: """Decorator to log execution time.""" def get_time_str(date_time) -> str: return time.strftime("%Y:%d:%m %H:%M:%S", time.gmtime(date_time)) extra_data = " | ".join([f'{k} - {v}' for k, v in _kwargs.items()]) def decorator(f): @wraps(f) def wrapper(*args, **kwargs): time_start = time.time() logger.info(f"{app_name} started at {get_time_str(time_start)} {extra_data}") result = f(*args, **kwargs) time_finish = time.time() logger.info(f"{app_name} finished at {get_time_str(time_finish)} {extra_data} " f"| total execution time in sec - {time_finish - time_start}") return result return wrapper return decorator def loop(logger, config: LoopRunningConfig): """Decorator to loop function continuously.""" def decorator(f): @wraps(f) def wrapper(*args, **kwargs): loop_num = 0 loop_settings = sorted(list(config.EXECUTION_TIME_DELAYS), key=lambda x: x[0]) while 1: loop_num += 1 time_start = time.time() time_logger(logger, config.APP_NAME, loop_num=loop_num)(f)(*args, **kwargs) execution_time = time.time() - time_start delay = 0 for constraint in loop_settings: execution_time_limit, delay = constraint if execution_time <= execution_time_limit: break logger.info(f"{config.APP_NAME} timeout - {delay}") time.sleep(delay) return wrapper return decorator