92 lines
4.0 KiB
Python
92 lines
4.0 KiB
Python
import mariadb
|
|
|
|
import logging
|
|
from typing import Generator
|
|
import datetime
|
|
|
|
from templating import SensorInfo
|
|
|
|
# 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 username_from_id(self, user_id):
|
|
self.cursor.execute("SELECT users.`username` FROM users WHERE `ID` = 1;")
|
|
return self.cursor.fetchone()
|
|
|
|
def view_valid_rooms(self, user_id) -> list[tuple[str, str, int]]:
|
|
self.cursor.execute(
|
|
"SELECT rooms.name, rooms.shortname, rooms.ID 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_shortname(self, shortname) -> list[int]:
|
|
self.cursor.execute("SELECT Sensors.`ID` from Sensors LEFT JOIN rooms ON Sensors.`roomID` = rooms.ID WHERE rooms.shortname = ?;",(shortname,))
|
|
return [x[0] for x in self.cursor.fetchall()]
|
|
|
|
def get_sensors_in_room_id(self, roomid) -> list[tuple[int, int]]:
|
|
self.cursor.execute("SELECT Sensors.`ID`, Sensors.`Type` from Sensors LEFT JOIN rooms ON Sensors.`roomID` = rooms.ID WHERE rooms.`ID` = ?;",(roomid,))
|
|
return [(x[0], x[1]) for x in self.cursor.fetchall()]
|
|
|
|
def get_sensorsinfo_by_sensorid(self, sensorid) -> SensorInfo | None:
|
|
self.cursor.execute("SELECT readings.`Timestamp`, readings.`reading` FROM readings WHERE readings.`sensorID` = ? ORDER BY readings.`Timestamp` DESC FETCH FIRST 1 ROWS ONLY;",(sensorid,))
|
|
fetch = self.cursor.fetchone()
|
|
if fetch is None:
|
|
return None
|
|
else:
|
|
return SensorInfo(0,sensorid, fetch[0], fetch[1])
|
|
|
|
|
|
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 Sensors ON types.`ID` = Sensors.`type` WHERE Sensors.`ID` = ?;",(sensor_ID,))
|
|
return self.cursor.fetchone()[0]
|
|
|
|
if __name__ == "__main__":
|
|
a = DatabaseConnect()
|
|
print(a.get_sensor_type(1)) |