diff --git a/src/backend/database.py b/src/backend/database.py index 32a6ec2..a6a0b26 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -3,16 +3,28 @@ import os import re import sqlite3 as sql import tempfile -from src.logic.log import MyLogger -from icecream import ic -from typing import List, Tuple, Dict, Any, Optional, Union from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union + +from icecream import ic from omegaconf import OmegaConf -from src.backend.db import CREATE_TABLE_APPARAT, CREATE_TABLE_MESSAGES, CREATE_TABLE_MEDIA, CREATE_TABLE_APPKONTOS, CREATE_TABLE_FILES, CREATE_TABLE_PROF, CREATE_TABLE_USER, CREATE_TABLE_SUBJECTS + +from src.backend.db import ( + CREATE_TABLE_APPARAT, + CREATE_TABLE_APPKONTOS, + CREATE_TABLE_FILES, + CREATE_TABLE_MEDIA, + CREATE_TABLE_MESSAGES, + CREATE_TABLE_PROF, + CREATE_TABLE_SUBJECTS, + CREATE_TABLE_USER, +) +from src.errors import AppPresentError, NoResultError from src.logic.constants import SEMAP_MEDIA_ACCOUNTS from src.logic.dataclass import ApparatData, BookData -from src.errors import NoResultError, AppPresentError -from src.utils import load_pickle, dump_pickle,create_blob +from src.logic.log import MyLogger +from src.utils import create_blob, dump_pickle, load_pickle + config = OmegaConf.load("config.yaml") logger = MyLogger(__name__) @@ -21,7 +33,8 @@ class Database: """ Initialize the database and create the tables if they do not exist. """ - def __init__(self, db_path: str = None): + + def __init__(self, db_path: str = "sap.db"): """ Default constructor for the database class @@ -36,9 +49,10 @@ class Database: if self.get_db_contents() is None: logger.log_critical("Database does not exist, creating tables") self.create_tables() - def get_db_contents(self)->Union[List[Tuple], None]: + + def get_db_contents(self) -> Union[List[Tuple], None]: """ - Get the contents of the + Get the contents of the Returns: Union[List[Tuple], None]: _description_ @@ -50,23 +64,27 @@ class Database: return cursor.fetchall() except sql.OperationalError: return None - def connect(self)->sql.Connection: + + def connect(self) -> sql.Connection: """ Connect to the database Returns: sql.Connection: The active connection to the database """ + print(self.db_path) return sql.connect(self.db_path) + def close_connection(self, conn: sql.Connection): """ closes the connection to the database - + Args: ---- - conn (sql.Connection): the connection to be closed """ conn.close() + def create_tables(self): """ Create the tables in the database @@ -83,7 +101,8 @@ class Database: cursor.execute(CREATE_TABLE_SUBJECTS) conn.commit() self.close_connection(conn) - def insertInto(self, query:str, params:Tuple) -> None: + + def insertInto(self, query: str, params: Tuple) -> None: """ Insert sent data into the database @@ -97,7 +116,10 @@ class Database: cursor.execute(query, params) conn.commit() self.close_connection(conn) - def query_db(self, query: str, args: Tuple = (), one: bool = False)->Union[Tuple, List[Tuple]]: + + def query_db( + self, query: str, args: Tuple = (), one: bool = False + ) -> Union[Tuple, List[Tuple]]: """ Query the Database for the sent query. @@ -119,7 +141,9 @@ class Database: return (rv[0] if rv else None) if one else rv # Books - def addBookToDatabase(self, bookdata:BookData,app_id:Union[str,int], prof_id:Union[str,int]): + def addBookToDatabase( + self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] + ): """ Add books to the database. Both app_id and prof_id are required to add the book to the database, as the app_id and prof_id are used to select the books later on. @@ -160,7 +184,10 @@ class Database: cursor.execute(query, params) conn.commit() self.close_connection(conn) - def getBookIdBasedOnSignature(self, app_id:Union[str,int], prof_id:Union[str,int],signature:str)->int: + + def getBookIdBasedOnSignature( + self, app_id: Union[str, int], prof_id: Union[str, int], signature: str + ) -> int: """ Get a book id based on the signature of the book. @@ -172,11 +199,17 @@ class Database: Returns: int: The id of the book """ - result = self.query_db("SELECT bookdata, id FROM media WHERE app_id=? AND prof_id=?", (app_id,prof_id)) - books = [(load_pickle(i[0]),i[1]) for i in result] + result = self.query_db( + "SELECT bookdata, id FROM media WHERE app_id=? AND prof_id=?", + (app_id, prof_id), + ) + books = [(load_pickle(i[0]), i[1]) for i in result] book = [i for i in books if i[0].signature == signature][0][1] return book - def getBookBasedOnSignature(self, app_id:Union[str,int], prof_id:Union[str,int],signature:str)->BookData: + + def getBookBasedOnSignature( + self, app_id: Union[str, int], prof_id: Union[str, int], signature: str + ) -> BookData: """ Get the book based on the signature of the book. @@ -188,11 +221,14 @@ class Database: Returns: BookData: The total metadata of the book wrapped in a BookData object """ - result = self.query_db("SELECT bookdata FROM media WHERE app_id=? AND prof_id=?", (app_id,prof_id)) + result = self.query_db( + "SELECT bookdata FROM media WHERE app_id=? AND prof_id=?", (app_id, prof_id) + ) books = [load_pickle(i[0]) for i in result] book = [i for i in books if i.signature == signature][0] return book - def getLastBookId(self)->int: + + def getLastBookId(self) -> int: """ Get the last book id in the database @@ -200,12 +236,13 @@ class Database: int: ID of the last book in the database """ return self.query_db("SELECT id FROM media ORDER BY id DESC", one=True)[0] - def searchBook(self, data:dict[str, str])->list[tuple[BookData, int]]: + + def searchBook(self, data: dict[str, str]) -> list[tuple[BookData, int]]: """ Search a book in the database based on the sent data. Args: - data (dict[str, str]): A dictionary containing the data to be searched for. The dictionary can contain the following: + data (dict[str, str]): A dictionary containing the data to be searched for. The dictionary can contain the following: - signature: The signature of the book - title: The title of the book @@ -215,7 +252,7 @@ class Database: rdata = self.query_db("SELECT * FROM media WHERE deleted=0") ic(rdata, len(rdata)) mode = 0 - if len(data)== 1: + if len(data) == 1: if "signature" in data.keys(): mode = 1 elif "title" in data.keys(): @@ -231,16 +268,20 @@ class Database: prof_id = book[3] if mode == 1: if data["signature"] in bookdata.signature: - ret.append((bookdata,app_id,prof_id)) + ret.append((bookdata, app_id, prof_id)) elif mode == 2: if data["title"] in bookdata.title: - ret.append((bookdata,app_id,prof_id)) + ret.append((bookdata, app_id, prof_id)) elif mode == 3: - if data["signature"] in bookdata.signature and data["title"] in bookdata.title: - ret.append((bookdata,app_id,prof_id)) + if ( + data["signature"] in bookdata.signature + and data["title"] in bookdata.title + ): + ret.append((bookdata, app_id, prof_id)) ic(ret) return ret - def setAvailability(self, book_id:str, available:str): + + def setAvailability(self, book_id: str, available: str): """ Set the availability of a book in the database @@ -248,8 +289,11 @@ class Database: book_id (str): The id of the book available (str): The availability of the book """ - self.query_db("UPDATE media SET available=? WHERE id=?", (available,book_id)) - def getBookId(self, bookdata:BookData, app_id:Union[str,int], prof_id:Union[str,int])->int: + self.query_db("UPDATE media SET available=? WHERE id=?", (available, book_id)) + + def getBookId( + self, bookdata: BookData, app_id: Union[str, int], prof_id: Union[str, int] + ) -> int: """ Get the id of a book based on the metadata of the book @@ -261,9 +305,14 @@ class Database: Returns: int: ID of the book """ - result = self.query_db("SELECT id FROM media WHERE bookdata=? AND app_id=? AND prof_id=?", (dump_pickle(bookdata),app_id,prof_id), one=True) + result = self.query_db( + "SELECT id FROM media WHERE bookdata=? AND app_id=? AND prof_id=?", + (dump_pickle(bookdata), app_id, prof_id), + one=True, + ) return result[0] - def getBook(self,book_id:int)->BookData: + + def getBook(self, book_id: int) -> BookData: """ Get the book based on the id in the database @@ -273,8 +322,15 @@ class Database: Returns: BookData: The metadata of the book wrapped in a BookData object """ - return load_pickle(self.query_db("SELECT bookdata FROM media WHERE id=?", (book_id,), one=True)[0]) - def getBooks(self, app_id:Union[str,int], prof_id:Union[str,int], deleted=0)->list[dict[int, BookData, int]]: + return load_pickle( + self.query_db( + "SELECT bookdata FROM media WHERE id=?", (book_id,), one=True + )[0] + ) + + def getBooks( + self, app_id: Union[str, int], prof_id: Union[str, int], deleted=0 + ) -> list[dict[int, BookData, int]]: """ Get the Books based on the apparat id and the professor id @@ -286,7 +342,9 @@ class Database: Returns: list[dict[int, BookData, int]]: A list of dictionaries containing the id, the metadata of the book and the availability of the book """ - qdata = self.query_db(f"SELECT id,bookdata,available FROM media WHERE (app_id={app_id} AND prof_id={prof_id}) AND (deleted={deleted if deleted == 0 else '1 OR deleted=0'})") + qdata = self.query_db( + f"SELECT id,bookdata,available FROM media WHERE (app_id={app_id} AND prof_id={prof_id}) AND (deleted={deleted if deleted == 0 else '1 OR deleted=0'})" + ) ret_result = [] for result_a in qdata: data = {"id": int, "bookdata": BookData, "available": int} @@ -295,7 +353,8 @@ class Database: data["available"] = result_a[2] ret_result.append(data) return ret_result - def updateBookdata(self, book_id, bookdata:BookData): + + def updateBookdata(self, book_id, bookdata: BookData): """ Update the bookdata in the database @@ -303,7 +362,10 @@ class Database: book_id (str): The id of the book bookdata (BookData): The new metadata of the book """ - self.query_db("UPDATE media SET bookdata=? WHERE id=?", (dump_pickle(bookdata),book_id)) + self.query_db( + "UPDATE media SET bookdata=? WHERE id=?", (dump_pickle(bookdata), book_id) + ) + def deleteBook(self, book_id): """ Delete a book from the database @@ -314,7 +376,7 @@ class Database: self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,)) # File Interactions - def getBlob(self, filename, app_id:Union[str,int]): + def getBlob(self, filename, app_id: Union[str, int]): """ Get a blob from the database @@ -323,10 +385,17 @@ class Database: app_id (str): ID of the apparat Returns: - bytes: The file stored in + bytes: The file stored in """ - return self.query_db("SELECT fileblob FROM files WHERE filename=? AND app_id=?", (filename,app_id), one=True)[0] - def insertFile(self, file: list[dict], app_id:Union[str,int], prof_id:Union[str,int]): + return self.query_db( + "SELECT fileblob FROM files WHERE filename=? AND app_id=?", + (filename, app_id), + one=True, + )[0] + + def insertFile( + self, file: list[dict], app_id: Union[str, int], prof_id: Union[str, int] + ): """Instert a list of files into the database Args: @@ -343,8 +412,11 @@ class Database: continue blob = create_blob(path) query = "INSERT OR IGNORE INTO files (filename, fileblob, app_id, filetyp,prof_id) VALUES (?, ?, ?, ?,?)" - self.query_db(query, (filename, blob, app_id, filetyp,prof_id)) - def recreateFile(self, filename:str, app_id:Union[str,int],filetype:str)->str: + self.query_db(query, (filename, blob, app_id, filetyp, prof_id)) + + def recreateFile( + self, filename: str, app_id: Union[str, int], filetype: str + ) -> str: """Recreate a file from the database Args: @@ -353,12 +425,12 @@ class Database: filetype (str): the extension of the file to be created Returns: - str: The filename of the recreated file + str: The filename of the recreated file """ blob = self.getBlob(filename, app_id) tempdir = config.database.tempdir tempdir = tempdir.replace("~", str(Path.home())) - tempdir_path = Path(tempdir) + tempdir_path = Path(tempdir) if not os.path.exists(tempdir_path): os.mkdir(tempdir_path) file = tempfile.NamedTemporaryFile( @@ -367,7 +439,8 @@ class Database: file.write(blob) print("file created") return file.name - def getFiles(self, app_id:Union[str,int], prof_id:int)->list[tuple]: + + def getFiles(self, app_id: Union[str, int], prof_id: int) -> list[tuple]: """Get all the files associated with the apparat and the professor Args: @@ -377,9 +450,12 @@ class Database: Returns: list[tuple]: a list of tuples containing the filename and the filetype for the corresponding apparat and professor """ - return self.query_db("SELECT filename, filetyp FROM files WHERE app_id=? AND prof_id=?", (app_id,prof_id)) + return self.query_db( + "SELECT filename, filetyp FROM files WHERE app_id=? AND prof_id=?", + (app_id, prof_id), + ) - def getSemersters(self)->list[str]: + def getSemersters(self) -> list[str]: """Return all the unique semesters in the database Returns: @@ -397,7 +473,7 @@ class Database: return self.query_db("SELECT * FROM subjects") # Messages - def addMessage(self, message:dict,user:str, app_id:Union[str,int]): + def addMessage(self, message: dict, user: str, app_id: Union[str, int]): """add a Message to the database Args: @@ -405,32 +481,40 @@ class Database: user (str): the user who added the message app_id (Union[str,int]): the id of the apparat """ + def __getUserId(user): - return self.query_db("SELECT id FROM user WHERE username=?", (user,), one=True)[0] + return self.query_db( + "SELECT id FROM user WHERE username=?", (user,), one=True + )[0] + user_id = __getUserId(user) - self.query_db("INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", (message["message"],user_id,message["remind_at"],app_id)) - def getMessages(self, date:str)->list[dict[str, str, str, str]]: + self.query_db( + "INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", + (message["message"], user_id, message["remind_at"], app_id), + ) + + def getMessages(self, date: str) -> list[dict[str, str, str, str]]: """Get all the messages for a specific date Args: date (str): a date.datetime object formatted as a string in the format "YYYY-MM-DD" - + Returns: list[dict[str, str, str, str]]: a list of dictionaries containing the message, the user who added the message, the apparat id and the id of the message """ + def __get_user_name(user_id): - return self.query_db("SELECT username FROM user WHERE id=?", (user_id,), one=True)[0] + return self.query_db( + "SELECT username FROM user WHERE id=?", (user_id,), one=True + )[0] + messages = self.query_db("SELECT * FROM messages WHERE remind_at=?", (date,)) ret = [ - { - "message": i[2], - "user": __get_user_name(i[4]), - "appnr": i[5], - "id": i[0] - } + {"message": i[2], "user": __get_user_name(i[4]), "appnr": i[5], "id": i[0]} for i in messages ] return ret + def deleteMessage(self, message_id): """Delete a message from the database @@ -440,7 +524,7 @@ class Database: self.query_db("DELETE FROM messages WHERE id=?", (message_id,)) # Prof data - def getProfNameById(self, prof_id:Union[str,int],add_title:bool=False)->str: + def getProfNameById(self, prof_id: Union[str, int], add_title: bool = False) -> str: """Get a professor name based on the id Args: @@ -450,12 +534,15 @@ class Database: Returns: str: The name of the professor """ - prof = self.query_db("SELECT fullname FROM prof WHERE id=?", (prof_id,), one=True) + prof = self.query_db( + "SELECT fullname FROM prof WHERE id=?", (prof_id,), one=True + ) if add_title: return f"{self.getTitleById(prof_id)}{prof[0]}" else: return prof[0] - def getTitleById(self, prof_id:Union[str,int])->str: + + def getTitleById(self, prof_id: Union[str, int]) -> str: """get the title of a professor based on the id Args: @@ -464,9 +551,12 @@ class Database: Returns: str: the title of the professor, with an added whitespace at the end, if no title is present, an empty string is returned """ - title = self.query_db("SELECT titel FROM prof WHERE id=?", (prof_id,), one=True)[0] + title = self.query_db( + "SELECT titel FROM prof WHERE id=?", (prof_id,), one=True + )[0] return f"{title} " if title is not None else "" - def getProfByName(self, prof_name:str)->tuple: + + def getProfByName(self, prof_name: str) -> tuple: """get all the data of a professor based on the name Args: @@ -475,8 +565,11 @@ class Database: Returns: tuple: the data of the professor """ - return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True) - def getProfId(self, prof_name:str)->Optional[int]: + return self.query_db( + "SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True + ) + + def getProfId(self, prof_name: str) -> Optional[int]: """Get the id of a professor based on the name Args: @@ -485,13 +578,14 @@ class Database: Returns: Optional[int]: the id of the professor, if the professor is not found, None is returned """ - + data = self.getProfByName(prof_name.replace(",", "")) if data is None: return None else: return data[0] - def getSpecificProfData(self, prof_id:Union[str,int], fields:List[str])->tuple: + + def getSpecificProfData(self, prof_id: Union[str, int], fields: List[str]) -> tuple: """A customisable function to get specific data of a professor based on the id Args: @@ -507,7 +601,8 @@ class Database: query = query[:-1] query += " FROM prof WHERE id=?" return self.query_db(query, (prof_id,), one=True)[0] - def getProfData(self, profname:str): + + def getProfData(self, profname: str): """Get mail, telephone number and title of a professor based on the name Args: @@ -516,9 +611,14 @@ class Database: Returns: tuple: the mail, telephone number and title of the professor """ - data = self.query_db("SELECT mail, telnr, titel FROM prof WHERE fullname=?", (profname.replace(",",""),), one=True) + data = self.query_db( + "SELECT mail, telnr, titel FROM prof WHERE fullname=?", + (profname.replace(",", ""),), + one=True, + ) return data - def createProf(self, prof_details:dict): + + def createProf(self, prof_details: dict): """Create a professor in the database Args: @@ -532,10 +632,18 @@ class Database: prof_fullname = prof_details["profname"].replace(",", "") prof_mail = prof_details["prof_mail"] prof_tel = prof_details["prof_tel"] - params = (prof_title, prof_fname, prof_lname, prof_mail, prof_tel, prof_fullname) + params = ( + prof_title, + prof_fname, + prof_lname, + prof_mail, + prof_tel, + prof_fullname, + ) query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)" self.insertInto(query=query, params=params) - def getProfs(self)->list[tuple]: + + def getProfs(self) -> list[tuple]: """Return all the professors in the database Returns: @@ -544,7 +652,7 @@ class Database: return self.query_db("SELECT * FROM prof") # Apparat - def getAllAparats(self,deleted=0)->list[tuple]: + def getAllAparats(self, deleted=0) -> list[tuple]: """Get all the apparats in the database Args: @@ -553,8 +661,11 @@ class Database: Returns: list[tuple]: a list of tuples containing the apparats """ - return self.query_db("SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,)) - def getApparatData(self, appnr, appname)->ApparatData: + return self.query_db( + "SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,) + ) + + def getApparatData(self, appnr, appname) -> ApparatData: """Get the Apparat data based on the apparat number and the name Args: @@ -567,7 +678,11 @@ class Database: Returns: ApparatData: the appended data of the apparat wrapped in an ApparatData object """ - result = self.query_db("SELECT * FROM semesterapparat WHERE appnr=? AND name=?", (appnr,appname), one=True) + result = self.query_db( + "SELECT * FROM semesterapparat WHERE appnr=? AND name=?", + (appnr, appname), + one=True, + ) if result is None: raise NoResultError("No result found") apparat = ApparatData() @@ -586,17 +701,21 @@ class Database: apparat.apparat_adis_id = result[11] apparat.prof_adis_id = result[12] return apparat - def getUnavailableApparatNumbers(self)->List[int]: + + def getUnavailableApparatNumbers(self) -> List[int]: """Get a list of all the apparat numbers in the database that are currently in use Returns: List[int]: the list of used apparat numbers """ - numbers = self.query_db("SELECT appnr FROM semesterapparat WHERE deletion_status=0") + numbers = self.query_db( + "SELECT appnr FROM semesterapparat WHERE deletion_status=0" + ) numbers = [i[0] for i in numbers] logger.log_info(f"Currently used apparat numbers: {numbers}") return numbers - def setNewSemesterDate(self, app_id:Union[str,int], newDate, dauerapp=False): + + def setNewSemesterDate(self, app_id: Union[str, int], newDate, dauerapp=False): """Set the new semester date for an apparat Args: @@ -606,10 +725,16 @@ class Database: """ date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d") if dauerapp: - self.query_db("UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", (date,dauerapp,app_id)) + self.query_db( + "UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", + (date, dauerapp, app_id), + ) else: - self.query_db("UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date,app_id)) - def getApparatId(self, apparat_name)->Optional[int]: + self.query_db( + "UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date, app_id) + ) + + def getApparatId(self, apparat_name) -> Optional[int]: """get the id of an apparat based on the name Args: @@ -618,12 +743,15 @@ class Database: Returns: Optional[int]: the id of the apparat, if the apparat is not found, None is returned """ - data = self.query_db("SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True) + data = self.query_db( + "SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True + ) if data is None: return None else: return data[0] - def createApparat(self, apparat:ApparatData)->int: + + def createApparat(self, apparat: ApparatData) -> int: """create the apparat in the database Args: @@ -635,7 +763,7 @@ class Database: Returns: Optional[int]: the id of the apparat """ - + prof_id = self.getProfId(apparat.profname) app_id = self.getApparatId(apparat.appname) if app_id: @@ -648,7 +776,8 @@ class Database: logger.log_info(query) self.query_db(query) return self.getApparatId(apparat.appname) - def getApparatsByProf(self, prof_id:Union[str,int])->list[tuple]: + + def getApparatsByProf(self, prof_id: Union[str, int]) -> list[tuple]: """Get all apparats based on the professor id Args: @@ -657,8 +786,11 @@ class Database: Returns: list[tuple]: a list of tuples containing the apparats """ - return self.query_db("SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,)) - def getApparatsBySemester(self, semester:str)->dict[list]: + return self.query_db( + "SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,) + ) + + def getApparatsBySemester(self, semester: str) -> dict[list]: """get all apparats based on the semester Args: @@ -667,7 +799,10 @@ class Database: Returns: dict[list]: a list off all created and deleted apparats for the selected semester """ - data = self.query_db("SELECT name, prof_id FROM semesterapparat WHERE erstellsemester=?", (semester,)) + data = self.query_db( + "SELECT name, prof_id FROM semesterapparat WHERE erstellsemester=?", + (semester,), + ) conn = self.connect() cursor = conn.cursor() c_tmp = [] @@ -695,7 +830,8 @@ class Database: d_ret[i[1]].append(i[0]) self.close_connection(conn) return {"created": c_ret, "deleted": d_ret} - def getApparatCountBySemester(self)->tuple[list[str],list[int]]: + + def getApparatCountBySemester(self) -> tuple[list[str], list[int]]: """get a list of all apparats created and deleted by semester Returns: @@ -724,15 +860,20 @@ class Database: ) ret.append(e_tuple) self.close_connection(conn) - return ret - def deleteApparat(self, app_id:Union[str,int], semester:str): + return ret + + def deleteApparat(self, app_id: Union[str, int], semester: str): """Delete an apparat from the database Args: app_id (Union[str, int]): the id of the apparat semester (str): the semester the apparat should be deleted from """ - self.query_db("UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", (semester,app_id)) + self.query_db( + "UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", + (semester, app_id), + ) + def isEternal(self, id): """check if the apparat is eternal (dauerapparat) @@ -742,8 +883,11 @@ class Database: Returns: int: the state of the apparat """ - return self.query_db("SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True) - def getApparatName(self, app_id:Union[str,int], prof_id:Union[str,int]): + return self.query_db( + "SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True + ) + + def getApparatName(self, app_id: Union[str, int], prof_id: Union[str, int]): """get the name of the apparat based on the id Args: @@ -753,8 +897,13 @@ class Database: Returns: str: the name of the apparat """ - return self.query_db("SELECT name FROM semesterapparat WHERE appnr=? AND prof_id=?", (app_id,prof_id), one=True)[0] - def updateApparat(self, apparat_data:ApparatData): + return self.query_db( + "SELECT name FROM semesterapparat WHERE appnr=? AND prof_id=?", + (app_id, prof_id), + one=True, + )[0] + + def updateApparat(self, apparat_data: ApparatData): """Update an apparat in the database Args: @@ -766,10 +915,11 @@ class Database: apparat_data.app_fach, apparat_data.dauerapp, self.getProfId(apparat_data.profname), - apparat_data.appnr, + apparat_data.appnr, ) self.query_db(query, params) - def checkApparatExists(self, apparat_name:str): + + def checkApparatExists(self, apparat_name: str): """check if the apparat is already present in the database based on the name Args: @@ -778,8 +928,17 @@ class Database: Returns: bool: True if the apparat is present, False if not """ - return True if self.query_db("SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True) else False - def checkApparatExistsById(self, app_id:Union[str,int])->bool: + return ( + True + if self.query_db( + "SELECT appnr FROM semesterapparat WHERE name=?", + (apparat_name,), + one=True, + ) + else False + ) + + def checkApparatExistsById(self, app_id: Union[str, int]) -> bool: """a check to see if the apparat is already present in the database, based on the id Args: @@ -788,11 +947,18 @@ class Database: Returns: bool: True if the apparat is present, False if not """ - return True if self.query_db("SELECT appnr FROM semesterapparat WHERE appnr=?", (app_id,), one=True) else False + return ( + True + if self.query_db( + "SELECT appnr FROM semesterapparat WHERE appnr=?", (app_id,), one=True + ) + else False + ) + # Statistics def statistic_request(self, **kwargs: Any): - """Take n amount of kwargs and return the result of the query - """ + """Take n amount of kwargs and return the result of the query""" + def __query(query): """execute the query and return the result @@ -815,6 +981,7 @@ class Database: result[result.index(orig_value)] = result_a self.close_connection(conn) return result + if "deletable" in kwargs.keys(): query = f"SELECT * FROM semesterapparat WHERE deletion_status=0 AND dauer=0 AND (erstellsemester!='{kwargs['deletesemester']}' OR verlängerung_bis!='{kwargs['deletesemester']}')" return __query(query) @@ -854,13 +1021,14 @@ class Database: def getUser(self): """Get a single user from the database""" return self.query_db("SELECT * FROM user", one=True) - def getUsers(self)->list[tuple]: + + def getUsers(self) -> list[tuple]: """Return a list of tuples of all the users in the database""" return self.query_db("SELECT * FROM user") def login(self, user, hashed_password): """try to login the user. - The salt for the user will be requested from the database and then added to the hashed password. The password will then be compared to the password in the database + The salt for the user will be requested from the database and then added to the hashed password. The password will then be compared to the password in the database Args: user (str): username that tries to login @@ -869,15 +1037,20 @@ class Database: Returns: bool: True if the login was successful, False if not """ - salt = self.query_db("SELECT salt FROM user WHERE username=?", (user,), one=True)[0] + salt = self.query_db( + "SELECT salt FROM user WHERE username=?", (user,), one=True + )[0] if salt is None: return False hashed_password = salt + hashed_password - password = self.query_db("SELECT password FROM user WHERE username=?", (user,), one=True)[0] + password = self.query_db( + "SELECT password FROM user WHERE username=?", (user,), one=True + )[0] if password == hashed_password: return True else: return False + def changePassword(self, user, new_password): """change the password of a user. The password will be added with the salt and then committed to the database @@ -886,9 +1059,14 @@ class Database: user (str): username new_password (str): the hashed password """ - salt = self.query_db("SELECT salt FROM user WHERE username=?", (user,), one=True)[0] + salt = self.query_db( + "SELECT salt FROM user WHERE username=?", (user,), one=True + )[0] new_password = salt + new_password - self.query_db("UPDATE user SET password=? WHERE username=?", (new_password,user)) + self.query_db( + "UPDATE user SET password=? WHERE username=?", (new_password, user) + ) + def getRole(self, user): """get the role of the user @@ -898,15 +1076,19 @@ class Database: Returns: str: the name of the role """ - return self.query_db("SELECT role FROM user WHERE username=?", (user,), one=True)[0] - def getRoles(self)->list[tuple]: + return self.query_db( + "SELECT role FROM user WHERE username=?", (user,), one=True + )[0] + + def getRoles(self) -> list[tuple]: """get all the roles in the database Returns: list[str]: a list of all the roles """ return self.query_db("SELECT role FROM user") - def checkUsername(self, user)->bool: + + def checkUsername(self, user) -> bool: """a check to see if the username is already present in the database Args: @@ -915,8 +1097,11 @@ class Database: Returns: bool: True if the username is present, False if not """ - data = self.query_db("SELECT username FROM user WHERE username=?", (user,), one=True) + data = self.query_db( + "SELECT username FROM user WHERE username=?", (user,), one=True + ) return True if data is not None else False + def createUser(self, user, password, role, salt): """create an user from the AdminCommands class. @@ -926,7 +1111,11 @@ class Database: role (str): the role of the user salt (str): a salt for the password """ - self.query_db("INSERT OR IGNORE INTO user (username, password, role, salt) VALUES (?,?,?,?)", (user,password,role,salt)) + self.query_db( + "INSERT OR IGNORE INTO user (username, password, role, salt) VALUES (?,?,?,?)", + (user, password, role, salt), + ) + def deleteUser(self, user): """delete an unser @@ -934,7 +1123,8 @@ class Database: user (str): username of the user """ self.query_db("DELETE FROM user WHERE username=?", (user,)) - def updateUser(self, username, data:dict[str, str]): + + def updateUser(self, username, data: dict[str, str]): """changge the data of a user Args: @@ -944,7 +1134,7 @@ class Database: conn = self.connect() cursor = conn.cursor() query = "UPDATE user SET " - for key,value in data.items(): + for key, value in data.items(): if key == "username": continue query += f"{key}='{value}'," @@ -954,7 +1144,8 @@ class Database: cursor.execute(query, params) conn.commit() self.close_connection(conn) - def getFacultyMember(self, name:str)->tuple: + + def getFacultyMember(self, name: str) -> tuple: """get a faculty member based on the name Args: @@ -963,8 +1154,13 @@ class Database: Returns: tuple: a tuple containing the data of the faculty member """ - return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", (name,), one=True) - def updateFacultyMember(self, data:dict, oldlname:str, oldfname:str): + return self.query_db( + "SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", + (name,), + one=True, + ) + + def updateFacultyMember(self, data: dict, oldlname: str, oldfname: str): """update the data of a faculty member Args: @@ -977,6 +1173,7 @@ class Database: data["oldlname"] = oldlname data["oldfname"] = oldfname self.query_db(query, data) + def getFacultyMembers(self): """get a list of all faculty members