HistorySQLite.read_node_history() - Code Metrics - Inspection of "Add SQLite history for Events" - FreeOpcUa/python-opcua - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#171)

unknown

created 2016-04-30 06:35 UTC

HistorySQLite.read_node_history() F

↳ Parent: HistorySQLite

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	33
CRAP Score	13.0976

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	13
c	3
b	0
f	0
dl	0
loc	54
ccs	33
cts	36
cp	0.9167
crap	13.0976
rs	3.5512

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._event_attributes = {}

        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
            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,)):

                    # rebuild the data value object
                    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, source_id, etype, period):

        with self._lock:
            _c_new = self._conn.cursor()

            table = self._get_table_name(source_id)

            # FIXME need to call vars(etype) to get all attributes as strings so proper table columns can be created
            # FIXME should a table already exist, but a new etype is supplied, table needs to be ALTERED

            self._datachanges_period[source_id] = period

            self._event_attributes[source_id] = etype.__dict__.keys()  # get fields (find a better way?)

            # create a table for the event which will store attributes of the Event 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,'
                               ' Timestamp TIMESTAMP,'
                               ' Time BLOB,'
                               ' ReceiveTime BLOB,'
                               ' LocalTime BLOB,'
                               ' EventId BLOB,'
                               ' EventType BLOB,'
                               ' Severity BLOB,'
                               ' Message BLOB,'
                               ' SourceName BLOB,'
                               ' SourceNode BLOB,'
                               ' ServerHandle BLOB)'.format(tn=table))

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

            self._conn.commit()

    def save_event(self, event):
        with self._lock:
            _c_sub = self._conn.cursor()

            table = self._get_table_name(event.SourceNode)

            placeholders, evtup = self._format_event(event)

            # insert the event into the database
            # print('INSERT INTO "{tn}" VALUES (NULL, "{ts}", '.format(tn=table, ts=event.Time) + placeholders + ')')
            try:
                _c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, "{ts}", {pl})'.format(tn=table, ts=event.Time, pl=placeholders), evtup)
            except sqlite3.Error as e:
                self.logger.error('Historizing SQL Insert Error for events from %s: %s', event.SourceNode, e)

            self._conn.commit()

            # get this node's period from the period dict and calculate the limit
            period = self._datachanges_period[event.SourceNode]

            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 Time < ?'.format(tn=table),
                                   (date_limit.isoformat(' '),))
                except sqlite3.Error as e:
                    self.logger.error('Historizing SQL Delete Old Data Error for events from %s: %s', event.SourceNode, e)

                self._conn.commit()

    def read_event_history(self, source_id, start, end, nb_values, evfilter):
        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(source_id)
            clauses = self._get_select_clauses(source_id, evfilter)

            cont = None
            results = []

            # print('SELECT {cl} FROM "{tn}" WHERE "Time" BETWEEN ? AND ? ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order))
            # test "EventId", "EventType", "SourceName", "Time", "Message", "Severity"

            # select events from the database; SQL select clause is built from EventFilter and available fields
            try:
                for row in _c_read.execute('SELECT {cl} FROM "{tn}" WHERE "Timestamp" BETWEEN ? AND ? '
                                           'ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order), (start_time, end_time, limit,)):

                    # place all the variants in the event field list object
                    hist_ev_field_list = ua.HistoryEventFieldList()
                    for field in row:
                        hist_ev_field_list.EventFields.append(ua.Variant.from_binary(Buffer(field)))

                    results.append(hist_ev_field_list)

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

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

                results = results[:nb_values]

            return results, cont

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

    def _format_event(self, event_result):
        placeholder = ''
        ev_variant_binaries = []

        ev_variants = event_result.get_fields()

        # convert the variants in each field to binary for storing in SQL BLOBs
        for variant in ev_variants:
            placeholder += '?, '
            ev_variant_binaries.append(variant.to_binary())

        placeholder = placeholder[:-2]  # remove trailing space and comma for SQL syntax
        evtup = tuple(ev_variant_binaries)

        return placeholder, evtup

    def _get_event_type_fields(self, etype):
        # FIXME finish and test
        etype_vars = vars(etype)
        etype_attr = etype_vars.keys()

    def _get_select_clauses(self, source_id, evfilter):
        s_clauses = []
        for select_clause in evfilter.SelectClauses:
            try:
                if not select_clause.BrowsePath:
                    s_clauses.append(select_clause.Attribute.name)
                else:
                    name = select_clause.BrowsePath[0].Name
                    s_clauses.append(name)
            except AttributeError:
                pass
                # FIXME what to do here?

        # remove select clauses that the event type doesn't have; SQL will error because the column doesn't exist
        clauses = [x for x in s_clauses if self._check(source_id, x)]
        clause_str = ''

        for select_clause in clauses:
            clause_str += '"' + select_clause + '", '

        return clause_str[:-2]  # remove trailing space and comma for SQL syntax

    def _check(self, source_id, s_clause):
        if s_clause in self._event_attributes[source_id]:
            return True
        else:
            return False

    def stop(self):
        with self._lock:
            self._conn.close()
            self.logger.info('Historizing SQL connection closed')


1	1		import logging
2	1		from datetime import timedelta
3	1		from datetime import datetime
4	1		from threading import Lock
5	1		from opcua import ua
6	1		from opcua.common.utils import Buffer
7	1		from opcua.server.history import HistoryStorageInterface
8	1		import sqlite3
9
10
11	1		class HistorySQLite(HistoryStorageInterface):
12			"""
13			very minimal history backend storing data in SQLite database
14			"""
15
16	1		def __init__(self, path="history.db"):
17	1		self.logger = logging.getLogger(__name__)
18	1		self._datachanges_period = {}
19	1		self._events = {}
20	1		self._db_file = path
21	1		self._lock = Lock()
22			self._event_attributes = {}
23	1
24			self._conn = sqlite3.connect(self._db_file, detect_types=sqlite3.PARSE_DECLTYPES, check_same_thread=False)
25	1
26	1	View Code Duplication	def new_historized_node(self, node_id, period, count=0):
			0 ignored issues – show Duplication introduced 2016-04-30 06:41 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
27	1		with self._lock:
28			_c_new = self._conn.cursor()
29	1
30			table = self._get_table_name(node_id)
31	1
32			self._datachanges_period[node_id] = period, count
33
34			# create a table for the node which will store attributes of the DataValue object
35	1		# note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
36	1		try:
37			_c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
38			' ServerTimestamp TIMESTAMP,'
39			' SourceTimestamp TIMESTAMP,'
40			' StatusCode INTEGER,'
41			' Value TEXT,'
42			' VariantType TEXT,'
43			' VariantBinary BLOB)'.format(tn=table))
44
45			except sqlite3.Error as e:
46			self.logger.info('Historizing SQL Table Creation Error for %s: %s', node_id, e)
47	1
48			self._conn.commit()
49	1
50	1		def save_node_value(self, node_id, datavalue):
51	1		with self._lock:
52			_c_sub = self._conn.cursor()
53	1
54			table = self._get_table_name(node_id)
55
56	1		# insert the data change into the database
57	1		try:
58			_c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, ?, ?, ?, ?, ?, ?)'.format(tn=table),
59			(
60			datavalue.ServerTimestamp,
61			datavalue.SourceTimestamp,
62			datavalue.StatusCode.value,
63			str(datavalue.Value.Value),
64			datavalue.Value.VariantType.name,
65			datavalue.Value.to_binary()
66			)
67			)
68			except sqlite3.Error as e:
69			self.logger.error('Historizing SQL Insert Error for %s: %s', node_id, e)
70	1
71			self._conn.commit()
72
73	1		# get this node's period from the period dict and calculate the limit
74			period, count = self._datachanges_period[node_id]
75	1
76			if period:
77			# after the insert, if a period was specified delete all records older than period
78			date_limit = datetime.now() - period
79
80			try:
81			_c_sub.execute('DELETE FROM "{tn}" WHERE ServerTimestamp < ?'.format(tn=table),
82			(date_limit.isoformat(' '),))
83			except sqlite3.Error as e:
84			self.logger.error('Historizing SQL Delete Old Data Error for %s: %s', node_id, e)
85
86			self._conn.commit()
87	1
88	1		def read_node_history(self, node_id, start, end, nb_values):
89	1		with self._lock:
90			_c_read = self._conn.cursor()
91	1
92			order = "ASC"
93	1
94	1		if start is None or start == ua.DateTimeMinValue:
95	1		order = "DESC"
96			start = ua.DateTimeMinValue
97	1
98	1		if end is None or end == ua.DateTimeMinValue:
99			end = datetime.utcnow() + timedelta(days=1)
100	1
101	1		if start < end:
102	1		start_time = start.isoformat(' ')
103			end_time = end.isoformat(' ')
104	1		else:
105	1		order = "DESC"
106	1		start_time = end.isoformat(' ')
107			end_time = start.isoformat(' ')
108	1
109	1		if nb_values:
110			limit = nb_values + 1 # add 1 to the number of values for retrieving a continuation point
111	1		else:
112			limit = -1 # in SQLite a LIMIT of -1 returns all results
113	1
114			table = self._get_table_name(node_id)
115	1
116	1		cont = None
117			results = []
118
119	1		# select values from the database; recreate UA Variant from binary
120	1		try:
121			for row in _c_read.execute('SELECT * FROM "{tn}" WHERE "ServerTimestamp" BETWEEN ? AND ? '
122	1		'ORDER BY "Id" {dir} LIMIT ?'.format(tn=table, dir=order), (start_time, end_time, limit,)):
123	1
124	1		# rebuild the data value object
125	1		dv = ua.DataValue(ua.Variant.from_binary(Buffer(row[6])))
126			dv.ServerTimestamp = row[1]
127	1		dv.SourceTimestamp = row[2]
128			dv.StatusCode = ua.StatusCode(row[3])
129
130			results.append(dv)
131
132	1		except sqlite3.Error as e:
133	1		self.logger.error('Historizing SQL Read Error for %s: %s', node_id, e)
134	1
135			if nb_values:
136	1		if start > ua.DateTimeMinValue and len(results) > nb_values:
137			cont = results[nb_values].ServerTimestamp
138	1
139			results = results[:nb_values]
140	1
141			return results, cont
142
143	1	View Code Duplication	def new_historized_event(self, source_id, etype, period):
			0 ignored issues – show Duplication introduced 2016-04-30 06:41 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
144			with self._lock:
145			_c_new = self._conn.cursor()
146	1
147			table = self._get_table_name(source_id)
148
149	1		# FIXME need to call vars(etype) to get all attributes as strings so proper table columns can be created
150	1		# FIXME should a table already exist, but a new etype is supplied, table needs to be ALTERED
151
152	1		self._datachanges_period[source_id] = period
153	1
154	1		self._event_attributes[source_id] = etype.__dict__.keys() # get fields (find a better way?)
155
156			# create a table for the event which will store attributes of the Event object
157			# note: Value/VariantType TEXT is only for human reading, the actual data is stored in VariantBinary column
158			try:
159			_c_new.execute('CREATE TABLE "{tn}" (Id INTEGER PRIMARY KEY NOT NULL,'
160			' Timestamp TIMESTAMP,'
161			' Time BLOB,'
162			' ReceiveTime BLOB,'
163			' LocalTime BLOB,'
164			' EventId BLOB,'
165			' EventType BLOB,'
166			' Severity BLOB,'
167			' Message BLOB,'
168			' SourceName BLOB,'
169			' SourceNode BLOB,'
170			' ServerHandle BLOB)'.format(tn=table))
171
172			except sqlite3.Error as e:
173			self.logger.info('Historizing SQL Table Creation Error for events from %s: %s', source_id, e)
174
175			self._conn.commit()
176
177			def save_event(self, event):
178			with self._lock:
179			_c_sub = self._conn.cursor()
180
181			table = self._get_table_name(event.SourceNode)
182
183			placeholders, evtup = self._format_event(event)
184
185			# insert the event into the database
186			# print('INSERT INTO "{tn}" VALUES (NULL, "{ts}", '.format(tn=table, ts=event.Time) + placeholders + ')')
187			try:
188			_c_sub.execute('INSERT INTO "{tn}" VALUES (NULL, "{ts}", {pl})'.format(tn=table, ts=event.Time, pl=placeholders), evtup)
189			except sqlite3.Error as e:
190			self.logger.error('Historizing SQL Insert Error for events from %s: %s', event.SourceNode, e)
191
192			self._conn.commit()
193
194			# get this node's period from the period dict and calculate the limit
195			period = self._datachanges_period[event.SourceNode]
196
197			if period:
198			# after the insert, if a period was specified delete all records older than period
199			date_limit = datetime.now() - period
200
201			try:
202			_c_sub.execute('DELETE FROM "{tn}" WHERE Time < ?'.format(tn=table),
203			(date_limit.isoformat(' '),))
204			except sqlite3.Error as e:
205			self.logger.error('Historizing SQL Delete Old Data Error for events from %s: %s', event.SourceNode, e)
206
207			self._conn.commit()
208
209			def read_event_history(self, source_id, start, end, nb_values, evfilter):
210			with self._lock:
211
212			_c_read = self._conn.cursor()
213
214			order = "ASC"
215
216			if start is None or start == ua.DateTimeMinValue:
217			order = "DESC"
218			start = ua.DateTimeMinValue
219
220			if end is None or end == ua.DateTimeMinValue:
221			end = datetime.utcnow() + timedelta(days=1)
222
223			if start < end:
224			start_time = start.isoformat(' ')
225			end_time = end.isoformat(' ')
226			else:
227			order = "DESC"
228			start_time = end.isoformat(' ')
229			end_time = start.isoformat(' ')
230
231			if nb_values:
232			limit = nb_values + 1 # add 1 to the number of values for retrieving a continuation point
233			else:
234			limit = -1 # in SQLite a LIMIT of -1 returns all results
235
236			table = self._get_table_name(source_id)
237			clauses = self._get_select_clauses(source_id, evfilter)
238
239			cont = None
240			results = []
241
242			# print('SELECT {cl} FROM "{tn}" WHERE "Time" BETWEEN ? AND ? ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order))
243			# test "EventId", "EventType", "SourceName", "Time", "Message", "Severity"
244
245			# select events from the database; SQL select clause is built from EventFilter and available fields
246			try:
247			for row in _c_read.execute('SELECT {cl} FROM "{tn}" WHERE "Timestamp" BETWEEN ? AND ? '
248			'ORDER BY "Id" {dir} LIMIT ?'.format(cl=clauses, tn=table, dir=order), (start_time, end_time, limit,)):
249
250			# place all the variants in the event field list object
251			hist_ev_field_list = ua.HistoryEventFieldList()
252			for field in row:
253			hist_ev_field_list.EventFields.append(ua.Variant.from_binary(Buffer(field)))
254
255			results.append(hist_ev_field_list)
256
257			except sqlite3.Error as e:
258			self.logger.error('Historizing SQL Read Error events for node %s: %s', source_id, e)
259
260			if nb_values:
261			if start > ua.DateTimeMinValue and len(results) > nb_values:
262			cont = results[nb_values].Time
263
264			results = results[:nb_values]
265
266			return results, cont
267
268			def _get_table_name(self, node_id):
269			return str(node_id.NamespaceIndex) + '_' + str(node_id.Identifier)
270
271			def _format_event(self, event_result):
272			placeholder = ''
273			ev_variant_binaries = []
274
275			ev_variants = event_result.get_fields()
276
277			# convert the variants in each field to binary for storing in SQL BLOBs
278			for variant in ev_variants:
279			placeholder += '?, '
280			ev_variant_binaries.append(variant.to_binary())
281
282			placeholder = placeholder[:-2] # remove trailing space and comma for SQL syntax
283			evtup = tuple(ev_variant_binaries)
284
285			return placeholder, evtup
286
287			def _get_event_type_fields(self, etype):
288			# FIXME finish and test
289			etype_vars = vars(etype)
290			etype_attr = etype_vars.keys()
291
292			def _get_select_clauses(self, source_id, evfilter):
293			s_clauses = []
294			for select_clause in evfilter.SelectClauses:
295			try:
296			if not select_clause.BrowsePath:
297			s_clauses.append(select_clause.Attribute.name)
298			else:
299			name = select_clause.BrowsePath[0].Name
300			s_clauses.append(name)
301			except AttributeError:
302			pass
303			# FIXME what to do here?
304
305			# remove select clauses that the event type doesn't have; SQL will error because the column doesn't exist
306			clauses = [x for x in s_clauses if self._check(source_id, x)]
307			clause_str = ''
308
309			for select_clause in clauses:
310			clause_str += '"' + select_clause + '", '
311
312			return clause_str[:-2] # remove trailing space and comma for SQL syntax
313
314			def _check(self, source_id, s_clause):
315			if s_clause in self._event_attributes[source_id]:
316			return True
317			else:
318			return False
319
320			def stop(self):
321			with self._lock:
322			self._conn.close()
323			self.logger.info('Historizing SQL connection closed')
324

FreeOpcUa / python-opcua

Pull Request — master (#171)

HistorySQLite.read_node_history() F

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like