migrate repository again

db_connect.py

@@ -0,0 +1,73 @@
+import mariadb
+
+import logging
+from typing import Generator
+import datetime
+
+# main database connector class
+# permission validation is not performed on this class
+# perform those higher on the call stack
+class DatabaseConnect:
+    def __init__(self) -> None:
+        with open("db_creds.csv","r") as f:
+            credentials = f.read().split(",")
+            try:
+                conn = mariadb.connect(
+                    user=credentials[0],
+                    password=credentials[1],
+                    host=credentials[2],
+                    port=int(credentials[3]),
+                    database=credentials[4]
+                )
+            except mariadb.Error as e:
+                logging.fatal(f"Error connecting to database: {e}")
+                return
+            self.cursor = conn.cursor()
+        super().__init__()
+
+    def create_user(self, username: str, pwd: int) -> None:
+        self.cursor.execute("INSERT INTO Users (username, pwd) VALUE (?,?)",(username, pwd))
+        logging.info(f"Created user {username}")    
+
+    def delete_user(self, id) -> None:
+        self.cursor.execute("DELETE FROM Users WHERE ID = ?",(id))
+        logging.info(f"Deleted user {id}")
+
+    def display_users(self) -> tuple[tuple[int, str, int]]:
+        self.cursor.execute("SELECT * FROM Users")
+        return self.cursor.fetchall()
+        
+    def view_valid_rooms(self, user_id) -> list[tuple[str, str]]:
+        self.cursor.execute(
+            "SELECT rooms.name, rooms.shortname from permissions INNER JOIN rooms ON permissions.`roomID` = rooms.`ID` WHERE permissions.`userID` = ? AND permissions.`view` = 1;",
+            (user_id,))
+        return self.cursor.fetchall()
+    
+    def user_has_room_perms(self, user_id, room_shortname) -> bool:
+        self.cursor.execute("SELECT permissions.`view` FROM permissions LEFT JOIN rooms ON permissions.`roomID` = rooms.ID WHERE rooms.shortname = ? AND `userID` = ?;",(room_shortname, user_id))
+        res = self.cursor.fetchone()
+        if res is None:
+            return False
+        else:
+            return res[0] == 1
+    
+    # def roomID_from_shortname(self, shortname):
+    #    self.cursor.execute("SELECT rooms.`ID` from rooms WHERE rooms.shortname = ?;",(shortname,))
+    #    return self.cursor.fetchone()
+
+    def get_sensors_in_room(self, shortname) -> list[int]:
+        self.cursor.execute("SELECT devices.`ID` from devices LEFT JOIN rooms ON devices.`roomID` = rooms.ID WHERE rooms.shortname = ?;",(shortname,))
+        return [x[0] for x in self.cursor.fetchall()]
+
+    def get_sensor_data(self, sensor_ID: int) -> Generator[tuple[datetime.datetime, float]]:
+        self.cursor.execute("SELECT Timestamp, reading FROM Readings WHERE `sensorID` = ? ORDER BY `Timestamp` DESC;",(sensor_ID,))
+        for row in self.cursor:
+            yield row
+    
+    def get_sensor_type(self, sensor_ID):
+        self.cursor.execute("SELECT types.`type_desc` from types LEFT JOIN devices ON types.`ID` = devices.`type` WHERE devices.`ID` = ?;",(sensor_ID,))
+        return self.cursor.fetchone()[0]
+
+if __name__ == "__main__":
+    a = DatabaseConnect()
+    print(a.get_sensor_type(1))