GlancesExport.load_conf() - Code Metrics - Inspection of "Generate documentation for Glances 2.8" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2b80fa...6ea077 )

by Nicolas

created 2017-01-21 13:48 UTC

GlancesExport.load_conf() D

↳ Parent: GlancesExport

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	8
c	1
b	0
f	0
dl	0
loc	36
rs	4

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2017 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""
I am your father...

...for all Glances exports IF.
"""

from glances.compat import NoOptionError, NoSectionError, iteritems, iterkeys
from glances.logger import logger


class GlancesExport(object):

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

    def __init__(self, config=None, args=None):
        """Init the export class."""
        # Export name (= module name without glances_)
        self.export_name = self.__class__.__module__[len('glances_'):]
        logger.debug("Init export module %s" % self.export_name)

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

        # By default export is disable
        # Had 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

    def exit(self):
        """Close the export module."""
        logger.debug("Finalise export interface %s" % self.export_name)

    def plugins_to_export(self):
        """Return the list of plugins to export."""
        return ['cpu',
                'percpu',
                'load',
                'mem',
                'memswap',
                'network',
                'diskio',
                'fs',
                'processcount',
                'ip',
                'system',
                'uptime',
                'sensors',
                'docker',
                'uptime']

    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 mandatories parameters to load
        :param options: a list of optionnals 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.critical("No {} configuration found".format(section))
            return False
        except NoOptionError as e:
            logger.critical("Error in the {} configuration ({})".format(section, e))
            return False

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

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

        return True

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

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

        tags: a comma separated list of 'key:value' pairs.
            Example: foo:bar,spam:eggs
        dtags: a dict of tags.
            Example: {'foo': 'bar', 'spam': 'eggs'}
        """
        dtags = {}
        if tags:
            try:
                dtags = 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)
                dtags = {}

        return dtags

    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.

        Be aware that CSV export overwrite this class and use a specific one.
        """
        if not self.export_enable:
            return False

        # Get all the stats & limits
        all_stats = stats.getAllExports()
        all_limits = stats.getAllLimits()
        # Get the plugins list
        plugins = stats.getAllPlugins()

        # Loop over available plugins
        for i, plugin in enumerate(plugins):
            if plugin in self.plugins_to_export():
                if isinstance(all_stats[i], dict):
                    all_stats[i].update(all_limits[i])
                elif isinstance(all_stats[i], list):
                    all_stats[i] += all_limits[i]
                else:
                    continue
                export_names, export_values = self.__build_export(all_stats[i])
                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):
                pre_key = '{}.'.format(stats[stats['key']])
            else:
                pre_key = ''
            # Walk through the dict
            for key, value in iteritems(stats):
                if isinstance(value, list):
                    try:
                        value = value[0]
                    except IndexError:
                        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


1			# -- coding: utf-8 --
2			#
3			# This file is part of Glances.
4			#
5			# Copyright (C) 2017 Nicolargo <[email protected]>
6			#
7			# Glances is free software; you can redistribute it and/or modify
8			# it under the terms of the GNU Lesser General Public License as published by
9			# the Free Software Foundation, either version 3 of the License, or
10			# (at your option) any later version.
11			#
12			# Glances is distributed in the hope that it will be useful,
13			# but WITHOUT ANY WARRANTY; without even the implied warranty of
14			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15			# GNU Lesser General Public License for more details.
16			#
17			# You should have received a copy of the GNU Lesser General Public License
18			# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20			"""
21			I am your father...
22
23			...for all Glances exports IF.
24			"""
25
26			from glances.compat import NoOptionError, NoSectionError, iteritems, iterkeys
27			from glances.logger import logger
28
29
30			class GlancesExport(object):
31
32			"""Main class for Glances export IF."""
33
34			def __init__(self, config=None, args=None):
35			"""Init the export class."""
36			# Export name (= module name without glances_)
37			self.export_name = self.__class__.__module__[len('glances_'):]
38			logger.debug("Init export module %s" % self.export_name)
39
40			# Init the config & args
41			self.config = config
42			self.args = args
43
44			# By default export is disable
45			# Had to be set to True in the __init__ class of child
46			self.export_enable = False
47
48			# Mandatory for (most of) the export module
49			self.host = None
50			self.port = None
51
52			def exit(self):
53			"""Close the export module."""
54			logger.debug("Finalise export interface %s" % self.export_name)
55
56			def plugins_to_export(self):
57			"""Return the list of plugins to export."""
58			return ['cpu',
59			'percpu',
60			'load',
61			'mem',
62			'memswap',
63			'network',
64			'diskio',
65			'fs',
66			'processcount',
67			'ip',
68			'system',
69			'uptime',
70			'sensors',
71			'docker',
72			'uptime']
73
74			def load_conf(self, section, mandatories=['host', 'port'], options=None):
75			"""Load the export <section> configuration in the Glances configuration file.
76
77			:param section: name of the export section to load
78			:param mandatories: a list of mandatories parameters to load
79			:param options: a list of optionnals parameters to load
80
81			:returns: Boolean -- True if section is found
82			"""
83			options = options or []
84
85			if self.config is None:
86			return False
87
88			# By default read the mandatory host:port items
89			try:
90			for opt in mandatories:
91			setattr(self, opt, self.config.get_value(section, opt))
92			except NoSectionError:
93			logger.critical("No {} configuration found".format(section))
94			return False
95			except NoOptionError as e:
96			logger.critical("Error in the {} configuration ({})".format(section, e))
97			return False
98
99			# Load options
100			for opt in options:
101			try:
102			setattr(self, opt, self.config.get_value(section, opt))
103			except NoOptionError:
104			pass
105
106			logger.debug("Load {} from the Glances configuration file".format(section))
107			logger.debug("{} parameters: {}".format(section, {opt: getattr(self, opt) for opt in mandatories + options}))
108
109			return True
110
111			def get_item_key(self, item):
112			"""Return the value of the item 'key'."""
113			try:
114			ret = item[item['key']]
115			except KeyError:
116			logger.error("No 'key' available in {}".format(item))
117			if isinstance(ret, list):
118			return ret[0]
119			else:
120			return ret
121
122			def parse_tags(self, tags):
123			"""Parse tags into a dict.
124
125			tags: a comma separated list of 'key:value' pairs.
126			Example: foo:bar,spam:eggs
127			dtags: a dict of tags.
128			Example: {'foo': 'bar', 'spam': 'eggs'}
129			"""
130			dtags = {}
131			if tags:
132			try:
133			dtags = dict([x.split(':') for x in tags.split(',')])
134			except ValueError:
135			# one of the 'key:value' pairs was missing
136			logger.info('Invalid tags passed: %s', tags)
137			dtags = {}
138
139			return dtags
140
141			def update(self, stats):
142			"""Update stats to a server.
143
144			The method builds two lists: names and values
145			and calls the export method to export the stats.
146
147			Be aware that CSV export overwrite this class and use a specific one.
148			"""
149			if not self.export_enable:
150			return False
151
152			# Get all the stats & limits
153			all_stats = stats.getAllExports()
154			all_limits = stats.getAllLimits()
155			# Get the plugins list
156			plugins = stats.getAllPlugins()
157
158			# Loop over available plugins
159			for i, plugin in enumerate(plugins):
160			if plugin in self.plugins_to_export():
161			if isinstance(all_stats[i], dict):
162			all_stats[i].update(all_limits[i])
163			elif isinstance(all_stats[i], list):
164			all_stats[i] += all_limits[i]
165			else:
166			continue
167			export_names, export_values = self.__build_export(all_stats[i])
168			self.export(plugin, export_names, export_values)
169
170			return True
171
172			def __build_export(self, stats):
173			"""Build the export lists."""
174			export_names = []
175			export_values = []
176
177			if isinstance(stats, dict):
178			# Stats is a dict
179			# Is there a key ?
180			if 'key' in iterkeys(stats):
181			pre_key = '{}.'.format(stats[stats['key']])
182			else:
183			pre_key = ''
184			# Walk through the dict
185			for key, value in iteritems(stats):
186			if isinstance(value, list):
187			try:
188			value = value[0]
189			except IndexError:
190			value = ''
191			if isinstance(value, dict):
192			item_names, item_values = self.__build_export(value)
193			item_names = [pre_key + key.lower() + str(i) for i in item_names]
194			export_names += item_names
195			export_values += item_values
196			else:
197			export_names.append(pre_key + key.lower())
198			export_values.append(value)
199			elif isinstance(stats, list):
200			# Stats is a list (of dict)
201			# Recursive loop through the list
202			for item in stats:
203			item_names, item_values = self.__build_export(item)
204			export_names += item_names
205			export_values += item_values
206			return export_names, export_values
207

nicolargo / glances

Push — master ( 2b80fa...6ea077 )

GlancesExport.load_conf() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like