# -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import logging import typing T = typing.TypeVar('T') logger = logging.getLogger("cloudlinux-summary") class StatisticsDict(dict): """ Special class to store all metrics before sending it """ # a few values which should be return if we can't collect statistic for something module DEFAULT_RESULTS = { "str": "-42", "int": -42, "float": -42.0, "str_list": ["-42"], "int_list": [-42], "int_dict": { "-42": -42, }, "float_dict": { "-42": -42.0, }, "str_dict": { "-42": "-42", }, # Special default result. # Only for collecting of statistics about client's rpm packages "rpm_stat_list": [], } def evaluate_safe( self, func: typing.Callable[..., T], log_message: str | None ) -> tuple[bool, T]: """ Method for catching any exceptions while calling the passed function, logging them and returning None and success flag if an exception has been encountered. Returns a formatted result of the function call otherwise. :param func: function that should be called :param log_message: if given, it will be logged as exception message. """ try: result = func() except Exception as err: if log_message: logger.exception( '%s. Exception: "%s"', log_message, err, ) return False, None return True, result def add_metric( self, func: typing.Callable[..., T], type_of_result: str, name_of_metric: str, log_message: str | None ) -> None: """ Calls `evaluate_safe` and saves the result from it in the metric dictionary. :param log_message: if given, it will be logged as exception message. See method `evaluate_safe` for more details. """ success, result = self.evaluate_safe( func, log_message, ) if not success: result = self.DEFAULT_RESULTS[type_of_result] self[name_of_metric] = result