| Server IP : 84.32.84.139 / Your IP : 216.73.217.69 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/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