HistorySQLite.read_node_history() - Code Metrics - Inspection of "SQL History storage" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#158)

by Olivier

created 2016-04-17 07:31 UTC

HistorySQLite.read_node_history() F

↳ Parent: HistorySQLite

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	13
dl	0
loc	52
rs	2.9976

How to fix Long Method Complexity

import logging
from datetime import timedelta
from datetime import datetime
from threading import Lock
from opcua import ua
from opcua.common.utils import Buffer
from opcua.server.history import HistoryStorageInterface
import sqlite3


class HistorySQLite(HistoryStorageInterface):
    """
    very minimal history backend storing data in SQLite database
    """

    def __init__(self, path="history.db"):
        self.logger = logging.getLogger(__name__)
        self._datachanges_period = {}
        self._events = {}
        self._db_file = path
        self._lock = Lock()

        self._conn = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES, check_same_thread=False)

    def new_historized_node(self, node_id, period, count=0):
        with self._lock:
            _c_new = self._conn.cursor()

            table = self._get_table_name(node_id)

            self._datachanges_period[node_id] = period, count

            # create a table for the node which will store attributes of the DataValue object
            # note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
            try:
                _c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
                               ' ServerTimestamp TIMESTAMP,'
                               ' SourceTimestamp TIMESTAMP,'
                               ' StatusCode INTEGER,'
                               ' Value TEXT,'
                               ' VariantType TEXT,'
                               ' VariantBinary BLOB)'.format(tn=table))

            except sqlite3.Error as e:
                self.logger.info('Historizing SQL Table Creation Error for %s: %s', node_id, e)

            self._conn.commit()

    def save_node_value(self, node_id, datavalue):
        with self._lock:
            _c_sub = self._conn.cursor()

            table = self._get_table_name(node_id)

            # insert the data change into the database
            try:
                _c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, ?, ?, ?, ?, ?, ?)'.format(tn=table),
                               (
                                   datavalue.ServerTimestamp,
                                   datavalue.SourceTimestamp,
                                   datavalue.StatusCode.value,
                                   str(datavalue.Value.Value),
                                   datavalue.Value.VariantType.name,
                                   datavalue.Value.to_binary()
                               )
                               )
            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Insert Error for %s: %s', node_id, e)

            self._conn.commit()

            # get this node's period from the period dict and calculate the limit
            period, count = self._datachanges_period[node_id]

            if period:
                # after the insert, if a period was specified delete all records older than period
                date_limit = datetime.now() - period

                try:
                    _c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=table),
                                   (date_limit.isoformat(' '),))
                except sqlite3.Error as e:
                    self.logger.error('Historizing SQL Delete Old Data Error for %s: %s', node_id, e)

                self._conn.commit()

    def read_node_history(self, node_id, start, end, nb_values):
        with self._lock:
            _c_read = self._conn.cursor()

            order = "ASC"

            if start is None or start == ua.DateTimeMinValue:
                order = "DESC"
                start = ua.DateTimeMinValue

            if end is None or end == ua.DateTimeMinValue:
                end = datetime.utcnow() + timedelta(days=1)

            if start < end:
                start_time = start.isoformat(' ')
                end_time = end.isoformat(' ')
            else:
                order = "DESC"
                start_time = end.isoformat(' ')
                end_time = start.isoformat(' ')

            if nb_values:
                limit = nb_values + 1  # add 1 to the number of values for retrieving a continuation point
            else:
                limit = -1  # in SQLite a LIMIT of -1 returns all results

            table = self._get_table_name(node_id)

            cont = None
            results = []

            # select values from the database; recreate UA Variant from binary ORDER BY "ServerTimestamp" DESC
            try:
                for row in _c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
                                           'ORDER BY "Id" {dir} LIMIT ?'.format(tn=table, dir=order), (start_time, end_time, limit,)):
                    dv = ua.DataValue(ua.Variant.from_binary(Buffer(row[6])))
                    dv.ServerTimestamp = row[1]
                    dv.SourceTimestamp = row[2]
                    dv.StatusCode = ua.StatusCode(row[3])

                    results.append(dv)

            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Read Error for %s: %s', node_id, e)

            if nb_values:
                if start > ua.DateTimeMinValue and len(results) > nb_values:
                    cont = results[nb_values].ServerTimestamp

                results = results[:nb_values]

            return results, cont

    def new_historized_event(self, event, period):
        raise NotImplementedError

    def save_event(self, event):
        raise NotImplementedError

    def read_event_history(self, start, end, evfilter):
        raise NotImplementedError

    def _get_table_name(self, node_id):
        return str(node_id.NamespaceIndex) + '_' + str(node_id.Identifier)

    def stop(self):
        with self._lock:
            self._conn.close()



1			import logging
2			from datetime import timedelta
3			from datetime import datetime
4			from threading import Lock
5			from opcua import ua
6			from opcua.common.utils import Buffer
7			from opcua.server.history import HistoryStorageInterface
8			import sqlite3
9
10
11			class HistorySQLite(HistoryStorageInterface):
12			"""
13			very minimal history backend storing data in SQLite database
14			"""
15
16			def __init__(self, path="history.db"):
17			self.logger = logging.getLogger(__name__)
18			self._datachanges_period = {}
19			self._events = {}
20			self._db_file = path
21			self._lock = Lock()
22
23			self._conn = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES, check_same_thread=False)
24
25			def new_historized_node(self, node_id, period, count=0):
26			with self._lock:
27			_c_new = self._conn.cursor()
28
29			table = self._get_table_name(node_id)
30
31			self._datachanges_period[node_id] = period, count
32
33			# create a table for the node which will store attributes of the DataValue object
34			# note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
35			try:
36			_c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
37			' ServerTimestamp TIMESTAMP,'
38			' SourceTimestamp TIMESTAMP,'
39			' StatusCode INTEGER,'
40			' Value TEXT,'
41			' VariantType TEXT,'
42			' VariantBinary BLOB)'.format(tn=table))
43
44			except sqlite3.Error as e:
45			self.logger.info('Historizing SQL Table Creation Error for %s: %s', node_id, e)
46
47			self._conn.commit()
48
49			def save_node_value(self, node_id, datavalue):
50			with self._lock:
51			_c_sub = self._conn.cursor()
52
53			table = self._get_table_name(node_id)
54
55			# insert the data change into the database
56			try:
57			_c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, ?, ?, ?, ?, ?, ?)'.format(tn=table),
58			(
59			datavalue.ServerTimestamp,
60			datavalue.SourceTimestamp,
61			datavalue.StatusCode.value,
62			str(datavalue.Value.Value),
63			datavalue.Value.VariantType.name,
64			datavalue.Value.to_binary()
65			)
66			)
67			except sqlite3.Error as e:
68			self.logger.error('Historizing SQL Insert Error for %s: %s', node_id, e)
69
70			self._conn.commit()
71
72			# get this node's period from the period dict and calculate the limit
73			period, count = self._datachanges_period[node_id]
74
75			if period:
76			# after the insert, if a period was specified delete all records older than period
77			date_limit = datetime.now() - period
78
79			try:
80			_c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=table),
81			(date_limit.isoformat(' '),))
82			except sqlite3.Error as e:
83			self.logger.error('Historizing SQL Delete Old Data Error for %s: %s', node_id, e)
84
85			self._conn.commit()
86
87			def read_node_history(self, node_id, start, end, nb_values):
88			with self._lock:
89			_c_read = self._conn.cursor()
90
91			order = "ASC"
92
93			if start is None or start == ua.DateTimeMinValue:
94			order = "DESC"
95			start = ua.DateTimeMinValue
96
97			if end is None or end == ua.DateTimeMinValue:
98			end = datetime.utcnow() + timedelta(days=1)
99
100			if start < end:
101			start_time = start.isoformat(' ')
102			end_time = end.isoformat(' ')
103			else:
104			order = "DESC"
105			start_time = end.isoformat(' ')
106			end_time = start.isoformat(' ')
107
108			if nb_values:
109			limit = nb_values + 1 # add 1 to the number of values for retrieving a continuation point
110			else:
111			limit = -1 # in SQLite a LIMIT of -1 returns all results
112
113			table = self._get_table_name(node_id)
114
115			cont = None
116			results = []
117
118			# select values from the database; recreate UA Variant from binary ORDER BY "ServerTimestamp" DESC
119			try:
120			for row in _c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
121			'ORDER BY "Id" {dir} LIMIT ?'.format(tn=table, dir=order), (start_time, end_time, limit,)):
122			dv = ua.DataValue(ua.Variant.from_binary(Buffer(row[6])))
123			dv.ServerTimestamp = row[1]
124			dv.SourceTimestamp = row[2]
125			dv.StatusCode = ua.StatusCode(row[3])
126
127			results.append(dv)
128
129			except sqlite3.Error as e:
130			self.logger.error('Historizing SQL Read Error for %s: %s', node_id, e)
131
132			if nb_values:
133			if start > ua.DateTimeMinValue and len(results) > nb_values:
134			cont = results[nb_values].ServerTimestamp
135
136			results = results[:nb_values]
137
138			return results, cont
139
140			def new_historized_event(self, event, period):
141			raise NotImplementedError
142
143			def save_event(self, event):
144			raise NotImplementedError
145
146			def read_event_history(self, start, end, evfilter):
147			raise NotImplementedError
148
149			def _get_table_name(self, node_id):
150			return str(node_id.NamespaceIndex) + '_' + str(node_id.Identifier)
151
152			def stop(self):
153			with self._lock:
154			self._conn.close()
155
156

FreeOpcUa / python-opcua

Pull Request — master (#158)

HistorySQLite.read_node_history() F

Complexity

Size

Duplication

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like