Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
start_time, end_time, order, limit = self._get_bounds(start, end, nb_values)
cont = None
results = []
# select values from the database; recreate UA Variant from binary
try:
async with self._db.execute(
f'SELECT * FROM "{table}" WHERE "SourceTimestamp" BETWEEN ? AND ? '
f'ORDER BY "_Id" {order} LIMIT ?', (start_time, end_time, limit,)) as cursor:
async for row in cursor:
# rebuild the data value object
dv = ua.DataValue(variant_from_binary(Buffer(row[6])))
dv.ServerTimestamp = row[1]
dv.SourceTimestamp = row[2]
dv.StatusCode = ua.StatusCode(row[3])
results.append(dv)
except aiosqlite.Error as e:
self.logger.error("Historizing SQL Read Error for %s: %s", node_id, e)
if nb_values:
if len(results) > nb_values:
cont = results[nb_values].SourceTimestamp
results = results[:nb_values]
return results, cont