| Server IP : 172.67.174.47 / Your IP : 216.73.216.153 Web Server : LiteSpeed System : Linux premium151.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64 User : tempvsty ( 647) PHP Version : 8.0.30 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /./proc/thread-self/root/././opt/cloudlinux/venv/lib/python3.11/site-packages/clsummary/ |
Upload File : |
# -*- 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