note-samples/python-route-endpoint/dbstore.py at main · blues/note-samples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
import sqlite3

measurementTableName = "measurements"
alertTableName = "alerts"
class dbstore():
    _connection = None
    _cursor = None
    def __init__(self,file="") -> None:
        self._file = file


    def addMeasurement(self, deviceId, timestamp, type, value, units):

        d = (deviceId, timestamp, type, value, units,)

        self._cursor.execute(f'INSERT or REPLACE into {measurementTableName} VALUES(?,?,?,?,?)', d)
        self._connection.commit()

    def addAlert(self, deviceId, timestamp, type, message):

        d = (deviceId, timestamp, type, message,)

        self._cursor.execute(f'INSERT or REPLACE into {alertTableName} VALUES(?,?,?,?)', d)
        self._connection.commit()


    def connect(self) -> None:
        if self._connection is not None:
            return

        self._connection = sqlite3.connect(self._file)
        self._cursor = self._connection.cursor()

    def close(self) -> None:
        if not self._connection:
            return

        self._connection.close()
        self._connection = None

    def createTables(self):
        self._createMeasurementDataTableIfNotExist()
        self._createAlertTableIfNotExist()

    def getAlerts(self, limit=None):
        query = f""" SELECT * FROM {alertTableName}
                     ORDER BY ROWID DESC """

        if limit and limit > 0:
            query = f"""{query}
                      LIMIT {limit}
                     """
        c = self._cursor.execute(query)

        rows = c.fetchall()
        val = []
        for r in rows:
            val.append({
                "deviceId": r[0],
                "timestamp": r[1],
                "type": r[2],
                "message": r[3]
            })

        return val


    def _createMeasurementDataTableIfNotExist(self):
        self._cursor.execute(f"SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{measurementTableName}';")
        isTable = self._cursor.fetchone()[0]==1

        if isTable:
            return

        self._cursor.execute(f"""CREATE TABLE
          {measurementTableName}(
              device_id STRING,
              timestamp STRING,
              type STRING,
              value FLOAT,
              units STRING
           );""")

        self._connection.commit()

    def _createAlertTableIfNotExist(self):
        self._cursor.execute(f"SELECT count(name) FROM sqlite_master WHERE type='table' AND name='{alertTableName}';")
        isTable = self._cursor.fetchone()[0]==1

        if isTable:
            return

        self._cursor.execute(f"""CREATE TABLE
          {alertTableName}(
              device_id STRING,
              timestamp STRING,
              type STRING,
              message STRING
           );""")