glances.stats.GlancesStats.getAllFieldsDescriptionAsDict() - 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

GlancesStats.getAllFieldsDescriptionAsDict() A

↳ Parent: glances.stats

Complexity

Conditions

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	2
dl	0
loc	6
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
#

"""The stats manager."""

import collections
import os
import pathlib
import sys
import threading
import traceback
from importlib import import_module

from glances.globals import exports_path, plugins_path, sys_path
from glances.logger import logger
from glances.timer import Counter


class GlancesStats:
    """This class stores, updates and gives stats."""

    # Script header constant
    header = "glances_"

    def __init__(self, config=None, args=None):
        # Set the config instance
        self.config = config

        # Set the argument instance
        self.args = args

        # Load plugins and exports modules
        self.first_export = True
        self.load_modules(self.args)

    def __getattr__(self, item):
        """Overwrite the getattr method in case of attribute is not found.

        The goal is to dynamically generate the following methods:
        - getPlugname(): return Plugname stat in JSON format
        - getViewsPlugname(): return views of the Plugname stat in JSON format
        """
        # Check if the attribute starts with 'get'
        if item.startswith('getViews'):

            # Get the plugin name
            plugname = item[len('getViews') :].lower()
            # Get the plugin instance
            plugin = self._plugins[plugname]
            if hasattr(plugin, 'get_json_views'):
                # The method get_json_views exist, return it
                return getattr(plugin, 'get_json_views')
            # The method get_views is not found for the plugin
            raise AttributeError(item)
        if item.startswith('get'):

            # Get the plugin name
            plugname = item[len('get') :].lower()
            # Get the plugin instance
            plugin = self._plugins[plugname]
            if hasattr(plugin, 'get_json'):
                # The method get_json exist, return it
                return getattr(plugin, 'get_json')
            # The method get_stats is not found for the plugin
            raise AttributeError(item)
        # Default behavior
        raise AttributeError(item)

    def load_modules(self, args):
        """Wrapper to load: plugins and export modules."""

        # Init the plugins dict
        # Active plugins dictionary
        self._plugins = collections.defaultdict(dict)
        # Load the plugins
        self.load_plugins(args=args)

        # Load addititional plugins
        self.load_additional_plugins(args=args, config=self.config)

        # Init the export modules dict
        # Active exporters dictionary
        self._exports = collections.defaultdict(dict)
        # All available exporters dictionary
        self._exports_all = collections.defaultdict(dict)
        # Load the export modules
        self.load_exports(args=args)

        # Restoring system path
        sys.path = sys_path

    def _load_plugin(self, plugin_path, args=None, config=None):
        """Load the plugin, init it and add to the _plugin dict."""
        # Load the plugin class
        try:
            # Import the plugin
            plugin = import_module('glances.plugins.' + plugin_path)
            # Init and add the plugin to the dictionary
            self._plugins[plugin_path] = plugin.PluginModel(args=args, config=config)
        except Exception as e:
            # If a plugin can not be loaded, display a critical message
            # on the console but do not crash
            logger.critical(f"Error while initializing the {plugin_path} plugin ({e})")
            logger.error(traceback.format_exc())
            # An error occurred, disable the plugin
            if args is not None:
                setattr(args, 'disable_' + plugin_path, False)
        else:
            # Manage the default status of the plugin (enable or disable)
            if args is not None:
                # If the all keys are set in the disable_plugin option then look in the enable_plugin option
                if getattr(args, 'disable_all', False):
                    logger.debug('%s => %s', plugin_path, getattr(args, 'enable_' + plugin_path, False))
                    setattr(args, 'disable_' + plugin_path, not getattr(args, 'enable_' + plugin_path, False))
                else:
                    setattr(args, 'disable_' + plugin_path, getattr(args, 'disable_' + plugin_path, False))

    def load_plugins(self, args=None):
        """Load all plugins in the 'plugins' folder."""
        start_duration = Counter()

        for item in os.listdir(plugins_path):
            if os.path.isdir(os.path.join(plugins_path, item)) and not item.startswith('__') and item != 'plugin':
                # Load the plugin
                start_duration.reset()
                self._load_plugin(os.path.basename(item), args=args, config=self.config)
                logger.debug(f"Plugin {item} started in {start_duration.get()} seconds")

        # Log plugins list
        logger.debug(f"Active plugins list: {self.getPluginsList()}")

    def load_additional_plugins(self, args=None, config=None):
        """Load additional plugins if defined"""

        def get_addl_plugins(self, plugin_path):
            """Get list of additional plugins"""
            _plugin_list = []
            for plugin in os.listdir(plugin_path):
                path = os.path.join(plugin_path, plugin)
                if os.path.isdir(path) and not path.startswith('__'):
                    # Poor man's walk_pkgs - can't use pkgutil as the module would be already imported here!
                    for fil in pathlib.Path(path).glob('*.py'):
                        if fil.is_file():
                            with open(fil) as fd:
                                if 'PluginModel' in fd.read():
                                    _plugin_list.append(plugin)
                                    break

            return _plugin_list

        path = None
        # Skip section check as implied by has_option
        if config and config.parser.has_option('global', 'plugin_dir'):
            path = config.parser['global']['plugin_dir']

        if args and 'plugin_dir' in args and args.plugin_dir:
            path = args.plugin_dir

        if path:
            # Get list before starting the counter
            _sys_path = sys.path
            start_duration = Counter()
            # Ensure that plugins can be found in plugin_dir
            sys.path.insert(0, path)
            for plugin in get_addl_plugins(self, path):
                if plugin in sys.modules:
                    logger.warn(f"Plugin {plugin} already in sys.modules, skipping (workaround: rename plugin)")
                else:
                    start_duration.reset()
                    try:
                        _mod_loaded = import_module(plugin + '.model')
                        self._plugins[plugin] = _mod_loaded.PluginModel(args=args, config=config)
                        logger.debug(f"Plugin {plugin} started in {start_duration.get()} seconds")
                    except Exception as e:
                        # If a plugin can not be loaded, display a critical message
                        # on the console but do not crash
                        logger.critical(f"Error while initializing the {plugin} plugin ({e})")
                        logger.error(traceback.format_exc())
                        # An error occurred, disable the plugin
                        if args:
                            setattr(args, 'disable_' + plugin, False)

            sys.path = _sys_path
            # Log plugins list
            logger.debug(f"Active additional plugins list: {self.getPluginsList()}")

    def load_exports(self, args=None):
        """Load all exporters in the 'exports' folder."""
        start_duration = Counter()

        if args is None:
            return False

        for item in os.listdir(exports_path):
            if os.path.isdir(os.path.join(exports_path, item)) and not item.startswith('__'):
                # Load the exporter
                start_duration.reset()
                if item.startswith('glances_'):
                    # Avoid circular loop when Glances exporter uses lib with same name
                    # Example: influxdb should be named to glances_influxdb
                    exporter_name = os.path.basename(item).split('glances_')[1]
                else:
                    exporter_name = os.path.basename(item)
                # Set the disable_<name> to False by default
                setattr(self.args, 'export_' + exporter_name, getattr(self.args, 'export_' + exporter_name, False))
                # We should import the module
                if getattr(self.args, 'export_' + exporter_name, False):
                    # Import the export module
                    export_module = import_module(item)
                    # Add the exporter instance to the active exporters dictionary
                    self._exports[exporter_name] = export_module.Export(args=args, config=self.config)
                    # Add the exporter instance to the available exporters dictionary
                    self._exports_all[exporter_name] = self._exports[exporter_name]
                else:
                    # Add the exporter name to the available exporters dictionary
                    self._exports_all[exporter_name] = exporter_name
                logger.debug(f"Exporter {exporter_name} started in {start_duration.get()} seconds")

        # Log plugins list
        logger.debug(f"Active exports modules list: {self.getExportsList()}")
        return True

    def getPluginsList(self, enable=True):
        """Return the plugins list.

        if enable is True, only return the active plugins (default)
        if enable is False, return all the plugins

        Return: list of plugin name
        """
        if enable:
            return [p for p in self._plugins if self._plugins[p].is_enabled()]
        return list(self._plugins)

    def getExportsList(self, enable=True):
        """Return the exports list.

        if enable is True, only return the active exporters (default)
        if enable is False, return all the exporters

        :return: list of export module names
        """
        if enable:
            return list(self._exports)
        return list(self._exports_all)

    def load_limits(self, config=None):
        """Load the stats limits (except the one in the exclude list)."""
        # For each plugins (enable or not), call the load_limits method
        for p in self.getPluginsList(enable=False):
            self._plugins[p].load_limits(config)

    def __update_plugin(self, p):
        """Update stats, history and views for the given plugin name p"""
        self._plugins[p].update()
        self._plugins[p].update_views()
        self._plugins[p].update_stats_history()

    def update(self):
        """Wrapper method to update all stats.

        Only called by standalone and server modes
        """
        # Start update of all enable plugins
        for p in self.getPluginsList(enable=True):
            self.__update_plugin(p)

    def export(self, input_stats=None):
        """Export all the stats.

        Each export module is ran in a dedicated thread.
        """
        if self.first_export:
            # Init fields description
            # Why ? Because some exports modules need to know what is exported (name, type...)
            for e in self.getExportsList():
                logger.debug(f"Init exported stats using the {e} module")
                # @TODO: is it a good idea to thread this call ?
                self._exports[e].init(input_stats)
            # In this first loop, data are not exported because some information are missing (rate for example)
            logger.debug("Do not export stats during the first iteration because some information are missing")
            self.first_export = False
            return False

        input_stats = input_stats or {}

        for e in self.getExportsList():
            logger.debug(f"Export stats using the {e} module")
            thread = threading.Thread(target=self._exports[e].update, args=(input_stats,))
            thread.start()

        return True

    def getAll(self):
        """Return all the stats (list).
        This method is called byt the XML/RPC API.
        It should return all the plugins (enable or not) because filtering can be done by the client.
        """
        return [self._plugins[p].get_raw() for p in self.getPluginsList(enable=False)]

    def getAllAsDict(self, plugin_list=None):
        """Return all the stats (as dict).
        This method is called by the RESTFul API.
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_raw() for p in plugin_list}

    def getAllFieldsDescription(self):
        """Return all fields description (as list)."""
        return [self._plugins[p].fields_description for p in self.getPluginsList(enable=False)]

    def getAllFieldsDescriptionAsDict(self, plugin_list=None):
        """Return all fields description (as dict)."""
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].fields_description for p in plugin_list}

    def getAllExports(self, plugin_list=None):
        """Return all the stats to be exported as a list.

        Default behavior is to export all the stat
        if plugin_list is provided (list), only export stats of given plugins
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return [self._plugins[p].get_export() for p in plugin_list]

    def getAllExportsAsDict(self, plugin_list=None):
        """Return all the stats to be exported as a dict.

        Default behavior is to export all the stat
        if plugin_list is provided (list), only export stats of given plugins
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_export() for p in plugin_list}

    def getAllLimits(self, plugin_list=None):
        """Return the plugins limits list.

        Default behavior is to export all the limits
        if plugin_list is provided, only export limits of given plugin (list)
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return [self._plugins[p].limits for p in plugin_list]

    def getAllLimitsAsDict(self, plugin_list=None):
        """Return all the stats limits (dict).

        Default behavior is to export all the limits
        if plugin_list is provided, only export limits of given plugin (list)
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].limits for p in plugin_list}

    def getAllViews(self, plugin_list=None):
        """Return the plugins views.
        This method is called byt the XML/RPC API.
        It should return all the plugins views (enable or not) because filtering can be done by the client.
        """
        if plugin_list is None:
            plugin_list = self.getPluginsList(enable=False)
        return [self._plugins[p].get_views() for p in plugin_list]

    def getAllViewsAsDict(self, plugin_list=None):
        """Return all the stats views (dict).
        This method is called by the RESTFul API.
        """
        if plugin_list is None:
            # All enabled plugins should be exported
            plugin_list = self.getPluginsList()
        return {p: self._plugins[p].get_views() for p in plugin_list}

    def get_plugin(self, plugin_name):
        """Return the plugin stats."""
        if plugin_name in self._plugins:
            return self._plugins[plugin_name]
        return None

    def get_plugin_view(self, plugin_name):
        """Return the plugin views."""
        if plugin_name in self._plugins:
            return self._plugins[plugin_name].get_views()
        return None

    def end(self):
        """End of the Glances stats."""
        # Close export modules
        for e in self._exports:
            self._exports[e].exit()
        # Close plugins
        for p in self._plugins:
            self._plugins[p].exit()


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		"""The stats manager."""
10
11		import collections
12		import os
13		import pathlib
14		import sys
15		import threading
16		import traceback
17		from importlib import import_module
18
19		from glances.globals import exports_path, plugins_path, sys_path
20		from glances.logger import logger
21		from glances.timer import Counter
22
23
24		class GlancesStats:
25		"""This class stores, updates and gives stats."""
26
27		# Script header constant
28		header = "glances_"
29
30		def __init__(self, config=None, args=None):
31		# Set the config instance
32		self.config = config
33
34		# Set the argument instance
35		self.args = args
36
37		# Load plugins and exports modules
38		self.first_export = True
39		self.load_modules(self.args)
40
41		def __getattr__(self, item):
42		"""Overwrite the getattr method in case of attribute is not found.
43
44		The goal is to dynamically generate the following methods:
45		- getPlugname(): return Plugname stat in JSON format
46		- getViewsPlugname(): return views of the Plugname stat in JSON format
47		"""
48		# Check if the attribute starts with 'get'
49	View Code Duplication	if item.startswith('getViews'):
		0 ignored issues – show Duplication introduced 2024-05-16 22:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
50		# Get the plugin name
51		plugname = item[len('getViews') :].lower()
52		# Get the plugin instance
53		plugin = self._plugins[plugname]
54		if hasattr(plugin, 'get_json_views'):
55		# The method get_json_views exist, return it
56		return getattr(plugin, 'get_json_views')
57		# The method get_views is not found for the plugin
58		raise AttributeError(item)
59	View Code Duplication	if item.startswith('get'):
		0 ignored issues – show Duplication introduced 2024-05-16 22:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
60		# Get the plugin name
61		plugname = item[len('get') :].lower()
62		# Get the plugin instance
63		plugin = self._plugins[plugname]
64		if hasattr(plugin, 'get_json'):
65		# The method get_json exist, return it
66		return getattr(plugin, 'get_json')
67		# The method get_stats is not found for the plugin
68		raise AttributeError(item)
69		# Default behavior
70		raise AttributeError(item)
71
72		def load_modules(self, args):
73		"""Wrapper to load: plugins and export modules."""
74
75		# Init the plugins dict
76		# Active plugins dictionary
77		self._plugins = collections.defaultdict(dict)
78		# Load the plugins
79		self.load_plugins(args=args)
80
81		# Load addititional plugins
82		self.load_additional_plugins(args=args, config=self.config)
83
84		# Init the export modules dict
85		# Active exporters dictionary
86		self._exports = collections.defaultdict(dict)
87		# All available exporters dictionary
88		self._exports_all = collections.defaultdict(dict)
89		# Load the export modules
90		self.load_exports(args=args)
91
92		# Restoring system path
93		sys.path = sys_path
94
95		def _load_plugin(self, plugin_path, args=None, config=None):
96		"""Load the plugin, init it and add to the _plugin dict."""
97		# Load the plugin class
98		try:
99		# Import the plugin
100		plugin = import_module('glances.plugins.' + plugin_path)
101		# Init and add the plugin to the dictionary
102		self._plugins[plugin_path] = plugin.PluginModel(args=args, config=config)
103		except Exception as e:
104		# If a plugin can not be loaded, display a critical message
105		# on the console but do not crash
106		logger.critical(f"Error while initializing the {plugin_path} plugin ({e})")
107		logger.error(traceback.format_exc())
108		# An error occurred, disable the plugin
109		if args is not None:
110		setattr(args, 'disable_' + plugin_path, False)
111		else:
112		# Manage the default status of the plugin (enable or disable)
113		if args is not None:
114		# If the all keys are set in the disable_plugin option then look in the enable_plugin option
115		if getattr(args, 'disable_all', False):
116		logger.debug('%s => %s', plugin_path, getattr(args, 'enable_' + plugin_path, False))
117		setattr(args, 'disable_' + plugin_path, not getattr(args, 'enable_' + plugin_path, False))
118		else:
119		setattr(args, 'disable_' + plugin_path, getattr(args, 'disable_' + plugin_path, False))
120
121		def load_plugins(self, args=None):
122		"""Load all plugins in the 'plugins' folder."""
123		start_duration = Counter()
124
125		for item in os.listdir(plugins_path):
126		if os.path.isdir(os.path.join(plugins_path, item)) and not item.startswith('__') and item != 'plugin':
127		# Load the plugin
128		start_duration.reset()
129		self._load_plugin(os.path.basename(item), args=args, config=self.config)
130		logger.debug(f"Plugin {item} started in {start_duration.get()} seconds")
131
132		# Log plugins list
133		logger.debug(f"Active plugins list: {self.getPluginsList()}")
134
135		def load_additional_plugins(self, args=None, config=None):
136		"""Load additional plugins if defined"""
137
138		def get_addl_plugins(self, plugin_path):
139		"""Get list of additional plugins"""
140		_plugin_list = []
141		for plugin in os.listdir(plugin_path):
142		path = os.path.join(plugin_path, plugin)
143		if os.path.isdir(path) and not path.startswith('__'):
144		# Poor man's walk_pkgs - can't use pkgutil as the module would be already imported here!
145		for fil in pathlib.Path(path).glob('*.py'):
146		if fil.is_file():
147		with open(fil) as fd:
148		if 'PluginModel' in fd.read():
149		_plugin_list.append(plugin)
150		break
151
152		return _plugin_list
153
154		path = None
155		# Skip section check as implied by has_option
156		if config and config.parser.has_option('global', 'plugin_dir'):
157		path = config.parser['global']['plugin_dir']
158
159		if args and 'plugin_dir' in args and args.plugin_dir:
160		path = args.plugin_dir
161
162		if path:
163		# Get list before starting the counter
164		_sys_path = sys.path
165		start_duration = Counter()
166		# Ensure that plugins can be found in plugin_dir
167		sys.path.insert(0, path)
168		for plugin in get_addl_plugins(self, path):
169		if plugin in sys.modules:
170		logger.warn(f"Plugin {plugin} already in sys.modules, skipping (workaround: rename plugin)")
171		else:
172		start_duration.reset()
173		try:
174		_mod_loaded = import_module(plugin + '.model')
175		self._plugins[plugin] = _mod_loaded.PluginModel(args=args, config=config)
176		logger.debug(f"Plugin {plugin} started in {start_duration.get()} seconds")
177		except Exception as e:
178		# If a plugin can not be loaded, display a critical message
179		# on the console but do not crash
180		logger.critical(f"Error while initializing the {plugin} plugin ({e})")
181		logger.error(traceback.format_exc())
182		# An error occurred, disable the plugin
183		if args:
184		setattr(args, 'disable_' + plugin, False)
185
186		sys.path = _sys_path
187		# Log plugins list
188		logger.debug(f"Active additional plugins list: {self.getPluginsList()}")
189
190		def load_exports(self, args=None):
191		"""Load all exporters in the 'exports' folder."""
192		start_duration = Counter()
193
194		if args is None:
195		return False
196
197		for item in os.listdir(exports_path):
198		if os.path.isdir(os.path.join(exports_path, item)) and not item.startswith('__'):
199		# Load the exporter
200		start_duration.reset()
201		if item.startswith('glances_'):
202		# Avoid circular loop when Glances exporter uses lib with same name
203		# Example: influxdb should be named to glances_influxdb
204		exporter_name = os.path.basename(item).split('glances_')[1]
205		else:
206		exporter_name = os.path.basename(item)
207		# Set the disable_<name> to False by default
208		setattr(self.args, 'export_' + exporter_name, getattr(self.args, 'export_' + exporter_name, False))
209		# We should import the module
210		if getattr(self.args, 'export_' + exporter_name, False):
211		# Import the export module
212		export_module = import_module(item)
213		# Add the exporter instance to the active exporters dictionary
214		self._exports[exporter_name] = export_module.Export(args=args, config=self.config)
215		# Add the exporter instance to the available exporters dictionary
216		self._exports_all[exporter_name] = self._exports[exporter_name]
217		else:
218		# Add the exporter name to the available exporters dictionary
219		self._exports_all[exporter_name] = exporter_name
220		logger.debug(f"Exporter {exporter_name} started in {start_duration.get()} seconds")
221
222		# Log plugins list
223		logger.debug(f"Active exports modules list: {self.getExportsList()}")
224		return True
225
226		def getPluginsList(self, enable=True):
227		"""Return the plugins list.
228
229		if enable is True, only return the active plugins (default)
230		if enable is False, return all the plugins
231
232		Return: list of plugin name
233		"""
234		if enable:
235		return [p for p in self._plugins if self._plugins[p].is_enabled()]
236		return list(self._plugins)
237
238		def getExportsList(self, enable=True):
239		"""Return the exports list.
240
241		if enable is True, only return the active exporters (default)
242		if enable is False, return all the exporters
243
244		:return: list of export module names
245		"""
246		if enable:
247		return list(self._exports)
248		return list(self._exports_all)
249
250		def load_limits(self, config=None):
251		"""Load the stats limits (except the one in the exclude list)."""
252		# For each plugins (enable or not), call the load_limits method
253		for p in self.getPluginsList(enable=False):
254		self._plugins[p].load_limits(config)
255
256		def __update_plugin(self, p):
257		"""Update stats, history and views for the given plugin name p"""
258		self._plugins[p].update()
259		self._plugins[p].update_views()
260		self._plugins[p].update_stats_history()
261
262		def update(self):
263		"""Wrapper method to update all stats.
264
265		Only called by standalone and server modes
266		"""
267		# Start update of all enable plugins
268		for p in self.getPluginsList(enable=True):
269		self.__update_plugin(p)
270
271		def export(self, input_stats=None):
272		"""Export all the stats.
273
274		Each export module is ran in a dedicated thread.
275		"""
276		if self.first_export:
277		# Init fields description
278		# Why ? Because some exports modules need to know what is exported (name, type...)
279		for e in self.getExportsList():
280		logger.debug(f"Init exported stats using the {e} module")
281		# @TODO: is it a good idea to thread this call ?
282		self._exports[e].init(input_stats)
283		# In this first loop, data are not exported because some information are missing (rate for example)
284		logger.debug("Do not export stats during the first iteration because some information are missing")
285		self.first_export = False
286		return False
287
288		input_stats = input_stats or {}
289
290		for e in self.getExportsList():
291		logger.debug(f"Export stats using the {e} module")
292		thread = threading.Thread(target=self._exports[e].update, args=(input_stats,))
293		thread.start()
294
295		return True
296
297		def getAll(self):
298		"""Return all the stats (list).
299		This method is called byt the XML/RPC API.
300		It should return all the plugins (enable or not) because filtering can be done by the client.
301		"""
302		return [self._plugins[p].get_raw() for p in self.getPluginsList(enable=False)]
303
304		def getAllAsDict(self, plugin_list=None):
305		"""Return all the stats (as dict).
306		This method is called by the RESTFul API.
307		"""
308		if plugin_list is None:
309		# All enabled plugins should be exported
310		plugin_list = self.getPluginsList()
311		return {p: self._plugins[p].get_raw() for p in plugin_list}
312
313		def getAllFieldsDescription(self):
314		"""Return all fields description (as list)."""
315		return [self._plugins[p].fields_description for p in self.getPluginsList(enable=False)]
316
317		def getAllFieldsDescriptionAsDict(self, plugin_list=None):
318		"""Return all fields description (as dict)."""
319		if plugin_list is None:
320		# All enabled plugins should be exported
321		plugin_list = self.getPluginsList()
322		return {p: self._plugins[p].fields_description for p in plugin_list}
323
324		def getAllExports(self, plugin_list=None):
325		"""Return all the stats to be exported as a list.
326
327		Default behavior is to export all the stat
328		if plugin_list is provided (list), only export stats of given plugins
329		"""
330		if plugin_list is None:
331		# All enabled plugins should be exported
332		plugin_list = self.getPluginsList()
333		return [self._plugins[p].get_export() for p in plugin_list]
334
335		def getAllExportsAsDict(self, plugin_list=None):
336		"""Return all the stats to be exported as a dict.
337
338		Default behavior is to export all the stat
339		if plugin_list is provided (list), only export stats of given plugins
340		"""
341		if plugin_list is None:
342		# All enabled plugins should be exported
343		plugin_list = self.getPluginsList()
344		return {p: self._plugins[p].get_export() for p in plugin_list}
345
346		def getAllLimits(self, plugin_list=None):
347		"""Return the plugins limits list.
348
349		Default behavior is to export all the limits
350		if plugin_list is provided, only export limits of given plugin (list)
351		"""
352		if plugin_list is None:
353		# All enabled plugins should be exported
354		plugin_list = self.getPluginsList()
355		return [self._plugins[p].limits for p in plugin_list]
356
357		def getAllLimitsAsDict(self, plugin_list=None):
358		"""Return all the stats limits (dict).
359
360		Default behavior is to export all the limits
361		if plugin_list is provided, only export limits of given plugin (list)
362		"""
363		if plugin_list is None:
364		# All enabled plugins should be exported
365		plugin_list = self.getPluginsList()
366		return {p: self._plugins[p].limits for p in plugin_list}
367
368		def getAllViews(self, plugin_list=None):
369		"""Return the plugins views.
370		This method is called byt the XML/RPC API.
371		It should return all the plugins views (enable or not) because filtering can be done by the client.
372		"""
373		if plugin_list is None:
374		plugin_list = self.getPluginsList(enable=False)
375		return [self._plugins[p].get_views() for p in plugin_list]
376
377		def getAllViewsAsDict(self, plugin_list=None):
378		"""Return all the stats views (dict).
379		This method is called by the RESTFul API.
380		"""
381		if plugin_list is None:
382		# All enabled plugins should be exported
383		plugin_list = self.getPluginsList()
384		return {p: self._plugins[p].get_views() for p in plugin_list}
385
386		def get_plugin(self, plugin_name):
387		"""Return the plugin stats."""
388		if plugin_name in self._plugins:
389		return self._plugins[plugin_name]
390		return None
391
392		def get_plugin_view(self, plugin_name):
393		"""Return the plugin views."""
394		if plugin_name in self._plugins:
395		return self._plugins[plugin_name].get_views()
396		return None
397
398		def end(self):
399		"""End of the Glances stats."""
400		# Close export modules
401		for e in self._exports:
402		self._exports[e].exit()
403		# Close plugins
404		for p in self._plugins:
405		self._plugins[p].exit()
406

nicolargo / glances

Push — develop ( 584fbb...c5954c )

GlancesStats.getAllFieldsDescriptionAsDict() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like