glances.exports.export.GlancesExport.init() - Code Metrics - Inspection of "[Discussion] FreeBSD - AMD64 Install: Package py39..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 584fbb...c5954c )

by Nicolas

created 2025-04-06 15:51 UTC

glances.exports.export.GlancesExport.init() A

↳ Parent: glances.exports.export

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	8
rs	10
c	0
b	0
f	0

#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""
I am your father...
...for all Glances exports IF.
"""

from glances.globals import NoOptionError, NoSectionError, iteritems, iterkeys, json_dumps
from glances.logger import logger
from glances.timer import Counter


class GlancesExport:
    """Main class for Glances export IF."""

    # List of non exportable internal plugins
    non_exportable_plugins = [
        'alert',
        'help',
        'plugin',
        'psutilversion',
        'version',
    ]

    def __init__(self, config=None, args=None):
        """Init the export class."""
        # Export name
        self.export_name = self.__class__.__module__
        logger.debug(f"Init export module {self.export_name}")

        # Init the config & args
        self.config = config
        self.args = args

        # By default export is disabled
        # Needs to be set to True in the __init__ class of child
        self.export_enable = False

        # Mandatory for (most of) the export module
        self.host = None
        self.port = None

        # Save last export list
        self._last_exported_list = None

        # Fields description
        self._fields_description = None

    def _log_result_decorator(fct):

        """Log (DEBUG) the result of the function fct."""

        def wrapper(*args, **kw):
            counter = Counter()
            ret = fct(*args, **kw)
            duration = counter.get()
            class_name = args[0].__class__.__name__
            class_module = args[0].__class__.__module__
            logger.debug(f"{class_name} {class_module} {fct.__name__} return {ret} in {duration} seconds")
            return ret

        return wrapper

    def exit(self):
        """Close the export module."""
        logger.debug(f"Finalise export interface {self.export_name}")

    def load_conf(self, section, mandatories=['host', 'port'], options=None):
        """Load the export <section> configuration in the Glances configuration file.

        :param section: name of the export section to load
        :param mandatories: a list of mandatory parameters to load
        :param options: a list of optional parameters to load

        :returns: Boolean -- True if section is found
        """
        options = options or []

        if self.config is None:
            return False

        # By default read the mandatory host:port items
        try:
            for opt in mandatories:
                setattr(self, opt, self.config.get_value(section, opt))
        except NoSectionError:
            logger.error(f"No {section} configuration found")
            return False
        except NoOptionError as e:
            logger.error(f"Error in the {section} configuration ({e})")
            return False

        # Load options
        for opt in options:
            try:
                setattr(self, opt, self.config.get_value(section, opt))
            except NoOptionError:
                pass

        logger.debug(f"Load {section} from the Glances configuration file")
        logger.debug(f"{section} parameters: { ({opt: getattr(self, opt) for opt in mandatories + options}) }")

        return True

    def get_item_key(self, item):
        """Return the value of the item 'key'."""
        ret = None
        try:
            ret = item[item['key']]
        except KeyError:
            logger.error(f"No 'key' available in {item}")
        if isinstance(ret, list):
            return ret[0]
        return ret

    def parse_tags(self, tags):
        """Parse tags into a dict.

        :param tags: a comma-separated list of 'key:value' pairs. Example: foo:bar,spam:eggs
        :return: a dict of tags. Example: {'foo': 'bar', 'spam': 'eggs'}
        """
        d_tags = {}
        if tags:
            try:
                d_tags = dict([x.split(':') for x in tags.split(',')])
            except ValueError:
                # one of the 'key:value' pairs was missing
                logger.info('Invalid tags passed: %s', tags)
                d_tags = {}

        return d_tags

    def plugins_to_export(self, stats):
        """Return the list of plugins to export.

        :param stats: the stats object
        :return: a list of plugins to export
        """
        return [p for p in stats.getPluginsList() if p not in self.non_exportable_plugins]

    def last_exported_list(self):
        """Return the list of plugins last exported."""
        return self._last_exported_list

    def init(self, stats):
        """Return fields description in order to init stats in a server."""
        if not self.export_enable:
            return False

        self._last_exported_list = self.plugins_to_export(stats)
        self._fields_description = stats.getAllFieldsDescriptionAsDict(plugin_list=self.last_exported_list())
        return self._fields_description

    def update(self, stats):
        """Update stats to a server.

        The method builds two lists: names and values and calls the export method to export the stats.

        Note: this class can be overwritten (for example in CSV and Graph).
        """
        if not self.export_enable:
            return False

        # Get all the stats & limits
        self._last_exported_list = self.plugins_to_export(stats)
        all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
        all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())

        # Loop over plugins to export
        for plugin in self.last_exported_list():
            if isinstance(all_stats[plugin], dict):
                all_stats[plugin].update(all_limits[plugin])
                # Remove the <plugin>_disable field
                all_stats[plugin].pop(f'{plugin}_disable', None)
            elif isinstance(all_stats[plugin], list):
                # TypeError: string indices must be integers (Network plugin) #1054
                for i in all_stats[plugin]:
                    i.update(all_limits[plugin])
            else:
                continue
            export_names, export_values = self.build_export(all_stats[plugin])
            self.export(plugin, export_names, export_values)

        return True

    def build_export(self, stats):
        """Build the export lists."""
        export_names = []
        export_values = []

        if isinstance(stats, dict):
            # Stats is a dict
            # Is there a key ?
            if 'key' in iterkeys(stats) and stats['key'] in iterkeys(stats):
                pre_key = '{}.'.format(stats[stats['key']])
            else:
                pre_key = ''
            # Walk through the dict
            for key, value in sorted(iteritems(stats)):
                if isinstance(value, bool):
                    value = json_dumps(value).decode()

                if isinstance(value, list):
                    value = ' '.join([str(v) for v in value])

                if isinstance(value, dict):
                    item_names, item_values = self.build_export(value)
                    item_names = [pre_key + key.lower() + str(i) for i in item_names]
                    export_names += item_names
                    export_values += item_values
                else:
                    export_names.append(pre_key + key.lower())
                    export_values.append(value)
        elif isinstance(stats, list):
            # Stats is a list (of dict)
            # Recursive loop through the list
            for item in stats:
                item_names, item_values = self.build_export(item)
                export_names += item_names
                export_values += item_values
        return export_names, export_values

    def export(self, name, columns, points):
        # This method should be implemented by each exporter
        pass


1		#
2		# This file is part of Glances.
3		#
4		# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
5		#
6		# SPDX-License-Identifier: LGPL-3.0-only
7		#
8
9		"""
10		I am your father...
11		...for all Glances exports IF.
12		"""
13
14		from glances.globals import NoOptionError, NoSectionError, iteritems, iterkeys, json_dumps
15		from glances.logger import logger
16		from glances.timer import Counter
17
18
19		class GlancesExport:
20		"""Main class for Glances export IF."""
21
22		# List of non exportable internal plugins
23		non_exportable_plugins = [
24		'alert',
25		'help',
26		'plugin',
27		'psutilversion',
28		'version',
29		]
30
31		def __init__(self, config=None, args=None):
32		"""Init the export class."""
33		# Export name
34		self.export_name = self.__class__.__module__
35		logger.debug(f"Init export module {self.export_name}")
36
37		# Init the config & args
38		self.config = config
39		self.args = args
40
41		# By default export is disabled
42		# Needs to be set to True in the __init__ class of child
43		self.export_enable = False
44
45		# Mandatory for (most of) the export module
46		self.host = None
47		self.port = None
48
49		# Save last export list
50		self._last_exported_list = None
51
52		# Fields description
53		self._fields_description = None
54
55	View Code Duplication	def _log_result_decorator(fct):
		0 ignored issues – show Duplication introduced 2024-05-02 16:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
56		"""Log (DEBUG) the result of the function fct."""
57
58		def wrapper(args, *kw):
59		counter = Counter()
60		ret = fct(args, *kw)
61		duration = counter.get()
62		class_name = args[0].__class__.__name__
63		class_module = args[0].__class__.__module__
64		logger.debug(f"{class_name} {class_module} {fct.__name__} return {ret} in {duration} seconds")
65		return ret
66
67		return wrapper
68
69		def exit(self):
70		"""Close the export module."""
71		logger.debug(f"Finalise export interface {self.export_name}")
72
73		def load_conf(self, section, mandatories=['host', 'port'], options=None):
74		"""Load the export <section> configuration in the Glances configuration file.
75
76		:param section: name of the export section to load
77		:param mandatories: a list of mandatory parameters to load
78		:param options: a list of optional parameters to load
79
80		:returns: Boolean -- True if section is found
81		"""
82		options = options or []
83
84		if self.config is None:
85		return False
86
87		# By default read the mandatory host:port items
88		try:
89		for opt in mandatories:
90		setattr(self, opt, self.config.get_value(section, opt))
91		except NoSectionError:
92		logger.error(f"No {section} configuration found")
93		return False
94		except NoOptionError as e:
95		logger.error(f"Error in the {section} configuration ({e})")
96		return False
97
98		# Load options
99		for opt in options:
100		try:
101		setattr(self, opt, self.config.get_value(section, opt))
102		except NoOptionError:
103		pass
104
105		logger.debug(f"Load {section} from the Glances configuration file")
106		logger.debug(f"{section} parameters: { ({opt: getattr(self, opt) for opt in mandatories + options}) }")
107
108		return True
109
110		def get_item_key(self, item):
111		"""Return the value of the item 'key'."""
112		ret = None
113		try:
114		ret = item[item['key']]
115		except KeyError:
116		logger.error(f"No 'key' available in {item}")
117		if isinstance(ret, list):
118		return ret[0]
119		return ret
120
121		def parse_tags(self, tags):
122		"""Parse tags into a dict.
123
124		:param tags: a comma-separated list of 'key:value' pairs. Example: foo:bar,spam:eggs
125		:return: a dict of tags. Example: {'foo': 'bar', 'spam': 'eggs'}
126		"""
127		d_tags = {}
128		if tags:
129		try:
130		d_tags = dict([x.split(':') for x in tags.split(',')])
131		except ValueError:
132		# one of the 'key:value' pairs was missing
133		logger.info('Invalid tags passed: %s', tags)
134		d_tags = {}
135
136		return d_tags
137
138		def plugins_to_export(self, stats):
139		"""Return the list of plugins to export.
140
141		:param stats: the stats object
142		:return: a list of plugins to export
143		"""
144		return [p for p in stats.getPluginsList() if p not in self.non_exportable_plugins]
145
146		def last_exported_list(self):
147		"""Return the list of plugins last exported."""
148		return self._last_exported_list
149
150		def init(self, stats):
151		"""Return fields description in order to init stats in a server."""
152		if not self.export_enable:
153		return False
154
155		self._last_exported_list = self.plugins_to_export(stats)
156		self._fields_description = stats.getAllFieldsDescriptionAsDict(plugin_list=self.last_exported_list())
157		return self._fields_description
158
159		def update(self, stats):
160		"""Update stats to a server.
161
162		The method builds two lists: names and values and calls the export method to export the stats.
163
164		Note: this class can be overwritten (for example in CSV and Graph).
165		"""
166		if not self.export_enable:
167		return False
168
169		# Get all the stats & limits
170		self._last_exported_list = self.plugins_to_export(stats)
171		all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
172		all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())
173
174		# Loop over plugins to export
175		for plugin in self.last_exported_list():
176		if isinstance(all_stats[plugin], dict):
177		all_stats[plugin].update(all_limits[plugin])
178		# Remove the <plugin>_disable field
179		all_stats[plugin].pop(f'{plugin}_disable', None)
180		elif isinstance(all_stats[plugin], list):
181		# TypeError: string indices must be integers (Network plugin) #1054
182		for i in all_stats[plugin]:
183		i.update(all_limits[plugin])
184		else:
185		continue
186		export_names, export_values = self.build_export(all_stats[plugin])
187		self.export(plugin, export_names, export_values)
188
189		return True
190
191		def build_export(self, stats):
192		"""Build the export lists."""
193		export_names = []
194		export_values = []
195
196		if isinstance(stats, dict):
197		# Stats is a dict
198		# Is there a key ?
199		if 'key' in iterkeys(stats) and stats['key'] in iterkeys(stats):
200		pre_key = '{}.'.format(stats[stats['key']])
201		else:
202		pre_key = ''
203		# Walk through the dict
204		for key, value in sorted(iteritems(stats)):
205		if isinstance(value, bool):
206		value = json_dumps(value).decode()
207
208		if isinstance(value, list):
209		value = ' '.join([str(v) for v in value])
210
211		if isinstance(value, dict):
212		item_names, item_values = self.build_export(value)
213		item_names = [pre_key + key.lower() + str(i) for i in item_names]
214		export_names += item_names
215		export_values += item_values
216		else:
217		export_names.append(pre_key + key.lower())
218		export_values.append(value)
219		elif isinstance(stats, list):
220		# Stats is a list (of dict)
221		# Recursive loop through the list
222		for item in stats:
223		item_names, item_values = self.build_export(item)
224		export_names += item_names
225		export_values += item_values
226		return export_names, export_values
227
228		def export(self, name, columns, points):
229		# This method should be implemented by each exporter
230		pass
231

nicolargo / glances

Push — develop ( 584fbb...c5954c )

glances.exports.export.GlancesExport.init() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like