import time from datetime import datetime from functools import wraps from typing import Any, Callable, Dict from client import service from exceptions import EnvironmentMisconfigured from logger import logger 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 get_datetime_string(date: str, as_str: bool = True): """Get datetime from date and return it as a string or datetime Args: date: iso format date as_str: return type selector Returns: datetime as a string (by default) or datetime (if as_str=False is passed) """ return datetime.fromisoformat(date).isoformat() if as_str else datetime.fromisoformat(date) def get_date(value: str) -> str: datetime_value = datetime.fromisoformat(value) return datetime_value.date().isoformat() def check_clients(): try: service.apollo service.dsp service.user_data except RuntimeError as err: logger.error(f"{err}") raise EnvironmentMisconfigured(err) def remove_milliseconds(value: str) -> str: try: value = value[: value.index(".")] except ValueError: pass return value def check_artist(track: Dict[str, Any]) -> bool: return bool(track.get("public_meta", {}).get("artists", []))