glances.exports.glances_duckdb.Export.__init__() - Code Metrics - Inspection of "version 4.4.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 73a01d...ef36eb )

by Nicolas

created 2025-11-02 16:07 UTC

glances.exports.glances_duckdb.Export.init() A

↳ Parent: glances.exports.glances_duckdb

Complexity

Conditions

Size

Total Lines	25
Code Lines	12

Duplication

Lines	25
Ratio	100 %

Importance

Changes

Metric	Value
cc	2
eloc	12
nop	3
dl	25
loc	25
rs	9.8
c	0
b	0
f	0

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

"""DuckDB interface class."""

import sys
import time
from datetime import datetime
from platform import node

import duckdb

from glances.exports.export import GlancesExport
from glances.logger import logger

# Define the type conversions for DuckDB
# https://duckdb.org/docs/stable/clients/python/conversion
convert_types = {
    'bool': 'BOOLEAN',
    'int': 'BIGINT',
    'float': 'DOUBLE',
    'str': 'VARCHAR',
    'tuple': 'VARCHAR',  # Store tuples as VARCHAR (comma-separated)
    'list': 'VARCHAR',  # Store lists as VARCHAR (comma-separated)
    'NoneType': 'VARCHAR',
}


class Export(GlancesExport):
    """This class manages the DuckDB export module."""

    def __init__(self, config=None, args=None):

        """Init the DuckDB export IF."""
        super().__init__(config=config, args=args)

        # Mandatory configuration keys (additional to host and port)
        self.db = None

        # Optional configuration keys
        self.user = None
        self.password = None
        self.hostname = None

        # Load the configuration file
        self.export_enable = self.load_conf(
            'duckdb', mandatories=['database'], options=['user', 'password', 'hostname']
        )
        if not self.export_enable:
            exit('Missing DuckDB config')

        # The hostname is always add as an identifier in the DuckDB table
        # so we can filter the stats by hostname
        self.hostname = self.hostname or node().split(".")[0]

        # Init the DuckDB client
        self.client = self.init()

    def init(self):
        """Init the connection to the DuckDB server."""
        if not self.export_enable:
            return None

        try:
            db = duckdb.connect(database=self.database)
        except Exception as e:
            logger.critical(f"Cannot connect to DuckDB {self.database} ({e})")
            sys.exit(2)
        else:
            logger.info(f"Stats will be exported to DuckDB: {self.database}")

        return db

    def normalize(self, value):
        # Nothing to do...
        if isinstance(value, list) and len(value) == 1 and value[0] in ['True', 'False']:
            return bool(value[0])
        return value

    def update(self, stats):
        """Update the DuckDB export module."""
        if not self.export_enable:
            return False

        # Get all the stats & limits
        # Current limitation with sensors and fs plugins because fields list is not the same
        self._last_exported_list = [p for p in self.plugins_to_export(stats) if p not in ['sensors', 'fs']]
        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():
            # Remove some fields
            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):
                for i in all_stats[plugin]:
                    i.update(all_limits[plugin])
                    # Remove the <plugin>_disable field
                    i.pop(f"{plugin}_disable", None)
            else:
                continue

            plugin_stats = all_stats[plugin]
            creation_list = []  # List used to create the DuckDB table
            values_list = []  # List of values to insert (list of lists, one list per row)
            if isinstance(plugin_stats, dict):
                # Create the list to create the table
                creation_list.append('time TIMETZ')
                creation_list.append('hostname_id VARCHAR')
                for key, value in plugin_stats.items():
                    creation_list.append(f"{key} {convert_types[type(self.normalize(value)).__name__]}")
                # Create the list of values to insert
                item_list = []
                item_list.append(self.normalize(datetime.now().replace(microsecond=0)))
                item_list.append(self.normalize(f"{self.hostname}"))
                item_list.extend([self.normalize(value) for value in plugin_stats.values()])
                values_list = [item_list]
            elif isinstance(plugin_stats, list) and len(plugin_stats) > 0 and 'key' in plugin_stats[0]:
                # Create the list to create the table
                creation_list.append('time TIMETZ')
                creation_list.append('hostname_id VARCHAR')
                creation_list.append('key_id VARCHAR')
                for key, value in plugin_stats[0].items():
                    creation_list.append(f"{key} {convert_types[type(self.normalize(value)).__name__]}")
                # Create the list of values to insert
                for plugin_item in plugin_stats:
                    item_list = []
                    item_list.append(self.normalize(datetime.now().replace(microsecond=0)))
                    item_list.append(self.normalize(f"{self.hostname}"))
                    item_list.append(self.normalize(f"{plugin_item.get('key')}"))
                    item_list.extend([self.normalize(value) for value in plugin_item.values()])
                    values_list.append(item_list)
            else:
                continue

            # Export stats to DuckDB
            self.export(plugin, creation_list, values_list)

        return True

    def export(self, plugin, creation_list, values_list):
        """Export the stats to the DuckDB server."""
        logger.debug(f"Export {plugin} stats to DuckDB")

        # Create the table if it does not exist
        table_list = [t[0] for t in self.client.sql("SHOW TABLES").fetchall()]
        if plugin not in table_list:
            # Execute the create table query
            create_query = f"""
CREATE TABLE {plugin} (
{', '.join(creation_list)}
);"""
            logger.debug(f"Create table: {create_query}")
            try:
                self.client.execute(create_query)
            except Exception as e:
                logger.error(f"Cannot create table {plugin}: {e}")
                return

        # Commit the changes
        self.client.commit()

        # Insert values into the table
        for values in values_list:
            insert_query = f"""
INSERT INTO {plugin} VALUES (
{', '.join(['?' for _ in values])}
);"""
            logger.debug(f"Insert values into table {plugin}: {values}")
            try:
                self.client.execute(insert_query, values)
            except Exception as e:
                logger.error(f"Cannot insert data into table {plugin}: {e}")

        # Commit the changes
        self.client.commit()

    def exit(self):
        """Close the DuckDB export module."""
        # Force last write
        self.client.commit()

        # Close the DuckDB client
        time.sleep(3)  # Wait a bit to ensure all data is written
        self.client.close()

        # Call the father method
        super().exit()


1		#
2		# This file is part of Glances.
3		#
4		# SPDX-FileCopyrightText: 2025 Nicolas Hennion <[email protected]>
5		#
6		# SPDX-License-Identifier: LGPL-3.0-only
7		#
8
9		"""DuckDB interface class."""
10
11		import sys
12		import time
13		from datetime import datetime
14		from platform import node
15
16		import duckdb
17
18		from glances.exports.export import GlancesExport
19		from glances.logger import logger
20
21		# Define the type conversions for DuckDB
22		# https://duckdb.org/docs/stable/clients/python/conversion
23		convert_types = {
24		'bool': 'BOOLEAN',
25		'int': 'BIGINT',
26		'float': 'DOUBLE',
27		'str': 'VARCHAR',
28		'tuple': 'VARCHAR', # Store tuples as VARCHAR (comma-separated)
29		'list': 'VARCHAR', # Store lists as VARCHAR (comma-separated)
30		'NoneType': 'VARCHAR',
31		}
32
33
34		class Export(GlancesExport):
35		"""This class manages the DuckDB export module."""
36
37	View Code Duplication	def __init__(self, config=None, args=None):
		0 ignored issues – show Duplication introduced 2016-04-28 14:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
38		"""Init the DuckDB export IF."""
39		super().__init__(config=config, args=args)
40
41		# Mandatory configuration keys (additional to host and port)
42		self.db = None
43
44		# Optional configuration keys
45		self.user = None
46		self.password = None
47		self.hostname = None
48
49		# Load the configuration file
50		self.export_enable = self.load_conf(
51		'duckdb', mandatories=['database'], options=['user', 'password', 'hostname']
52		)
53		if not self.export_enable:
54		exit('Missing DuckDB config')
55
56		# The hostname is always add as an identifier in the DuckDB table
57		# so we can filter the stats by hostname
58		self.hostname = self.hostname or node().split(".")[0]
59
60		# Init the DuckDB client
61		self.client = self.init()
62
63		def init(self):
64		"""Init the connection to the DuckDB server."""
65		if not self.export_enable:
66		return None
67
68		try:
69		db = duckdb.connect(database=self.database)
70		except Exception as e:
71		logger.critical(f"Cannot connect to DuckDB {self.database} ({e})")
72		sys.exit(2)
73		else:
74		logger.info(f"Stats will be exported to DuckDB: {self.database}")
75
76		return db
77
78		def normalize(self, value):
79		# Nothing to do...
80		if isinstance(value, list) and len(value) == 1 and value[0] in ['True', 'False']:
81		return bool(value[0])
82		return value
83
84		def update(self, stats):
85		"""Update the DuckDB export module."""
86		if not self.export_enable:
87		return False
88
89		# Get all the stats & limits
90		# Current limitation with sensors and fs plugins because fields list is not the same
91		self._last_exported_list = [p for p in self.plugins_to_export(stats) if p not in ['sensors', 'fs']]
92		all_stats = stats.getAllExportsAsDict(plugin_list=self.last_exported_list())
93		all_limits = stats.getAllLimitsAsDict(plugin_list=self.last_exported_list())
94
95		# Loop over plugins to export
96		for plugin in self.last_exported_list():
97		# Remove some fields
98	View Code Duplication	if isinstance(all_stats[plugin], dict):
		0 ignored issues – show Duplication introduced 2025-09-21 15:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
99		all_stats[plugin].update(all_limits[plugin])
100		# Remove the <plugin>_disable field
101		all_stats[plugin].pop(f"{plugin}_disable", None)
102		elif isinstance(all_stats[plugin], list):
103		for i in all_stats[plugin]:
104		i.update(all_limits[plugin])
105		# Remove the <plugin>_disable field
106		i.pop(f"{plugin}_disable", None)
107		else:
108		continue
109
110		plugin_stats = all_stats[plugin]
111		creation_list = [] # List used to create the DuckDB table
112		values_list = [] # List of values to insert (list of lists, one list per row)
113		if isinstance(plugin_stats, dict):
114		# Create the list to create the table
115		creation_list.append('time TIMETZ')
116		creation_list.append('hostname_id VARCHAR')
117		for key, value in plugin_stats.items():
118		creation_list.append(f"{key} {convert_types[type(self.normalize(value)).__name__]}")
119		# Create the list of values to insert
120		item_list = []
121		item_list.append(self.normalize(datetime.now().replace(microsecond=0)))
122		item_list.append(self.normalize(f"{self.hostname}"))
123		item_list.extend([self.normalize(value) for value in plugin_stats.values()])
124		values_list = [item_list]
125		elif isinstance(plugin_stats, list) and len(plugin_stats) > 0 and 'key' in plugin_stats[0]:
126		# Create the list to create the table
127		creation_list.append('time TIMETZ')
128		creation_list.append('hostname_id VARCHAR')
129		creation_list.append('key_id VARCHAR')
130		for key, value in plugin_stats[0].items():
131		creation_list.append(f"{key} {convert_types[type(self.normalize(value)).__name__]}")
132		# Create the list of values to insert
133		for plugin_item in plugin_stats:
134		item_list = []
135		item_list.append(self.normalize(datetime.now().replace(microsecond=0)))
136		item_list.append(self.normalize(f"{self.hostname}"))
137		item_list.append(self.normalize(f"{plugin_item.get('key')}"))
138		item_list.extend([self.normalize(value) for value in plugin_item.values()])
139		values_list.append(item_list)
140		else:
141		continue
142
143		# Export stats to DuckDB
144		self.export(plugin, creation_list, values_list)
145
146		return True
147
148		def export(self, plugin, creation_list, values_list):
149		"""Export the stats to the DuckDB server."""
150		logger.debug(f"Export {plugin} stats to DuckDB")
151
152		# Create the table if it does not exist
153		table_list = [t[0] for t in self.client.sql("SHOW TABLES").fetchall()]
154		if plugin not in table_list:
155		# Execute the create table query
156		create_query = f"""
157		CREATE TABLE {plugin} (
158		{', '.join(creation_list)}
159		);"""
160		logger.debug(f"Create table: {create_query}")
161		try:
162		self.client.execute(create_query)
163		except Exception as e:
164		logger.error(f"Cannot create table {plugin}: {e}")
165		return
166
167		# Commit the changes
168		self.client.commit()
169
170		# Insert values into the table
171		for values in values_list:
172		insert_query = f"""
173		INSERT INTO {plugin} VALUES (
174		{', '.join(['?' for _ in values])}
175		);"""
176		logger.debug(f"Insert values into table {plugin}: {values}")
177		try:
178		self.client.execute(insert_query, values)
179		except Exception as e:
180		logger.error(f"Cannot insert data into table {plugin}: {e}")
181
182		# Commit the changes
183		self.client.commit()
184
185		def exit(self):
186		"""Close the DuckDB export module."""
187		# Force last write
188		self.client.commit()
189
190		# Close the DuckDB client
191		time.sleep(3) # Wait a bit to ensure all data is written
192		self.client.close()
193
194		# Call the father method
195		super().exit()
196

nicolargo / glances

Push — master ( 73a01d...ef36eb )

glances.exports.glances_duckdb.Export.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.exports.glances_duckdb.Export.init() A