PHB/SemesterapparatsManager
2
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

add wfh files

Browse Source
This commit is contained in:
WorldTeacher
2024-02-19 18:12:19 +01:00
parent 506206a0fb
commit 6d1119783f
13 changed files with 6618 additions and 1491 deletions
Download Patch File Download Diff File Expand all files Collapse all files

431
src/backend/database.py
View File

@@ -36,7 +36,6 @@ 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]:
"""
Get the contents of the
@@ -51,7 +50,6 @@ class Database:
return cursor.fetchall()
except sql.OperationalError:
return None
def connect(self)->sql.Connection:
"""
Connect to the database
@@ -60,7 +58,6 @@ class Database:
sql.Connection: The active connection to the database
"""
return sql.connect(self.db_path)
def close_connection(self, conn: sql.Connection):
"""
closes the connection to the database
@@ -70,7 +67,6 @@ class Database:
- conn (sql.Connection): the connection to be closed
"""
conn.close()
def create_tables(self):
"""
Create the tables in the database
@@ -87,7 +83,6 @@ class Database:
cursor.execute(CREATE_TABLE_SUBJECTS)
conn.commit()
self.close_connection(conn)
def insertInto(self, query:str, params:Tuple) -> None:
"""
Insert sent data into the database
@@ -102,7 +97,6 @@ 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]]:
"""
Query the Database for the sent query.
@@ -125,7 +119,7 @@ class Database:
return (rv[0] if rv else None) if one else rv
# Books
def addBookToDatabase(self, bookdata:BookData,app_id:str, prof_id:str):
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.
@@ -166,7 +160,7 @@ class Database:
cursor.execute(query, params)
conn.commit()
self.close_connection(conn)
def getBookIdBasedOnSignature(self, app_id:str, prof_id:str,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.
@@ -182,7 +176,7 @@ class Database:
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:str, prof_id:str,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.
@@ -255,7 +249,7 @@ class Database:
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, prof_id)->int:
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
@@ -280,7 +274,7 @@ class Database:
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, prof_id, deleted=0)->list[dict[int, BookData, int]]:
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
@@ -301,7 +295,6 @@ class Database:
data["available"] = result_a[2]
ret_result.append(data)
return ret_result
def updateBookdata(self, book_id, bookdata:BookData):
"""
Update the bookdata in the database
@@ -321,7 +314,7 @@ class Database:
self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,))
# File Interactions
def getBlob(self, filename, app_id):
def getBlob(self, filename, app_id:Union[str,int]):
"""
Get a blob from the database
@@ -333,7 +326,15 @@ class Database:
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: int, prof_id):
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:
file (list[dict]): a list containing all the files to be inserted
Structured: [{"name": "filename", "path": "path", "type": "filetype"}]
app_id (int): the id of the apparat
prof_id (str): the id of the professor
"""
for f in file:
filename = f["name"]
path = f["path"]
@@ -343,7 +344,17 @@ class Database:
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, app_id,filetype):
def recreateFile(self, filename:str, app_id:Union[str,int],filetype:str)->str:
"""Recreate a file from the database
Args:
filename (str): the name of the file
app_id (Union[str,int]): the id of the apparat
filetype (str): the extension of the file to be created
Returns:
str: The filename of the recreated file
"""
blob = self.getBlob(filename, app_id)
tempdir = config.database.tempdir
tempdir = tempdir.replace("~", str(Path.home()))
@@ -356,23 +367,57 @@ class Database:
file.write(blob)
print("file created")
return file.name
def getFiles(self, app_id: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:
app_id (Union[str,int]): The id of the apparat
prof_id (Union[str,int]): the id of the professor
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))
def getSemersters(self):
def getSemersters(self)->list[str]:
"""Return all the unique semesters in the database
Returns:
list: a list of strings containing the semesters
"""
data = self.query_db("SELECT DISTINCT erstellsemester FROM semesterapparat")
return [i[0] for i in data]
def getSubjects(self):
"""Get all the subjects in the database
Returns:
list[tuple]: a list of tuples containing the subjects
"""
return self.query_db("SELECT * FROM subjects")
# Messages
def addMessage(self, message:dict,user, appnr):
def addMessage(self, message:dict,user:str, app_id:Union[str,int]):
"""add a Message to the database
Args:
message (dict): the message to be added
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]
user_id = __getUserId(user)
self.query_db("INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", (message["message"],user_id,message["remind_at"],appnr))
def getMessages(self, date: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]
messages = self.query_db("SELECT * FROM messages WHERE remind_at=?", (date,))
@@ -387,48 +432,74 @@ class Database:
]
return ret
def deleteMessage(self, message_id):
"""Delete a message from the database
Args:
message_id (str): the id of the message
"""
self.query_db("DELETE FROM messages WHERE id=?", (message_id,))
# Prof data
def getProfNameById(self, prof_id:int,add_title:bool=False):
def getProfNameById(self, prof_id:Union[str,int],add_title:bool=False)->str:
"""Get a professor name based on the id
Args:
prof_id (Union[str,int]): The id of the professor
add_title (bool, optional): wether to add the title or no. Defaults to False.
Returns:
str: The name of the professor
"""
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:int):
def getTitleById(self, prof_id:Union[str,int])->str:
"""get the title of a professor based on the id
Args:
prof_id (Union[str,int]): the id of the professor
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]
return f"{title} " if title is not None else ""
def getProfByName(self, prof_name:str):
return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True)
def getProfId(self, prof_name:str):
"""
getProfId _summary_
def getProfByName(self, prof_name:str)->tuple:
"""get all the data of a professor based on the name
:param prof_name: _description_
:type prof_name: str
:return: _description_
:rtype: _type_
Args:
prof_name (str): the name of the professor
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]:
"""Get the id of a professor based on the name
Args:
prof_name (str): the name of the professor
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:int, fields:List[str]):
"""
getSpecificProfData _summary_
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:
----
- prof_id (int): _description_
- fields (List[str]): _description_
prof_id (Union[str,int]): the id of the professor
fields (List[str]): a list of fields to be returned
Returns:
-------
- _type_: _description_
tuple: a tuple containing the requested data
"""
query = "SELECT "
for field in fields:
@@ -437,10 +508,22 @@ class Database:
query += " FROM prof WHERE id=?"
return self.query_db(query, (prof_id,), one=True)[0]
def getProfData(self, profname:str):
"""Get mail, telephone number and title of a professor based on the name
Args:
profname (str): name of the professor
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)
return data
def createProf(self, prof_details:dict):
"""Create a professor in the database
Args:
prof_details (dict): a dictionary containing the details of the professor
"""
prof_title = prof_details["prof_title"]
prof_fname = prof_details["profname"].split(",")[1]
prof_fname = prof_fname.strip()
@@ -452,14 +535,38 @@ class Database:
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):
def getProfs(self)->list[tuple]:
"""Return all the professors in the database
Returns:
list[tuple]: a list containing all the professors in individual tuples
"""
return self.query_db("SELECT * FROM prof")
# Apparat
def getAllAparats(self,deleted=0):
def getAllAparats(self,deleted=0)->list[tuple]:
"""Get all the apparats in the database
Args:
deleted (int, optional): Switch the result to use . Defaults to 0.
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:
"""Get the Apparat data based on the apparat number and the name
Args:
appnr (str): the apparat number
appname (str): the name of the apparat
Raises:
NoResultError: an error is raised if no result is found
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)
if result is None:
raise NoResultError("No result found")
@@ -480,36 +587,59 @@ class Database:
apparat.prof_adis_id = result[12]
return apparat
def getUnavailableApparatNumbers(self)->List[int]:
"""
getUnavailableApparatNumbers returns a list of all currently used ApparatNumbers
"""Get a list of all the apparat numbers in the database that are currently in use
Returns:
-------
- number(List[int]): a list of all currently used apparat numbers
List[int]: the list of used apparat numbers
"""
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, appnr, newDate, dauerapp=False):
def setNewSemesterDate(self, app_id:Union[str,int], newDate, dauerapp=False):
"""Set the new semester date for an apparat
Args:
app_id (Union[str,int]): the id of the apparat
newDate (str): the new date
dauerapp (bool, optional): if the apparat was changed to dauerapparat. Defaults to False.
"""
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,appnr))
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,appnr))
def getApparatId(self, apparat_name):
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:
apparat_name (str): the name of the apparat e.g. "Semesterapparat 1"
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)
if data is None:
return None
else:
return data[0]
def createApparat(self, apparat:ApparatData)->Optional[AppPresentError]|int:
def createApparat(self, apparat:ApparatData)->int:
"""create the apparat in the database
Args:
apparat (ApparatData): the wrapped metadata of the apparat
Raises:
AppPresentError: an error describing that the apparats chosen id is already present in the database
Returns:
Optional[int]: the id of the apparat
"""
prof_id = self.getProfId(apparat.profname)
app_id = self.getApparatId(apparat.appname)
if app_id:
return AppPresentError(app_id)
raise AppPresentError(app_id)
self.createProf(apparat.get_prof_details())
prof_id = self.getProfId(apparat.profname)
@@ -518,9 +648,25 @@ class Database:
logger.log_info(query)
self.query_db(query)
return self.getApparatId(apparat.appname)
def getApparatsByProf(self, prof_id:int)->list[tuple]:
def getApparatsByProf(self, prof_id:Union[str,int])->list[tuple]:
"""Get all apparats based on the professor id
Args:
prof_id (Union[str,int]): the id of the professor
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:
def getApparatsBySemester(self, semester:str)->dict[list]:
"""get all apparats based on the semester
Args:
semester (str): the selected semester
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,))
conn = self.connect()
cursor = conn.cursor()
@@ -550,6 +696,11 @@ class Database:
self.close_connection(conn)
return {"created": c_ret, "deleted": d_ret}
def getApparatCountBySemester(self)->tuple[list[str],list[int]]:
"""get a list of all apparats created and deleted by semester
Returns:
tuple[list[str],list[int]]: a tuple containing two lists, the first list contains the semesters, the second list contains the amount of apparats created and deleted for the corresponding semester
"""
conn = self.connect()
cursor = conn.cursor()
semesters = self.getSemersters()
@@ -574,13 +725,41 @@ class Database:
ret.append(e_tuple)
self.close_connection(conn)
return ret
def deleteApparat(self, appnr, semester):
self.query_db("UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", (semester,appnr))
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))
def isEternal(self, id):
"""check if the apparat is eternal (dauerapparat)
Args:
id (int): the id of the apparat to be checked
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, prof_id):
def getApparatName(self, app_id:Union[str,int], prof_id:Union[str,int]):
"""get the name of the apparat based on the id
Args:
app_id (Union[str,int]): the id of the apparat
prof_id (Union[str,int]): the id of the professor
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):
"""Update an apparat in the database
Args:
apparat_data (ApparatData): the new metadata of the apparat
"""
query = f"UPDATE semesterapparat SET name = ?, fach = ?, dauer = ?, prof_id = ? WHERE appnr = ?"
params = (
apparat_data.appname,
@@ -590,13 +769,39 @@ class Database:
apparat_data.appnr,
)
self.query_db(query, params)
def checkApparatExists(self, apparat_name):
def checkApparatExists(self, apparat_name:str):
"""check if the apparat is already present in the database based on the name
Args:
apparat_name (str): the name of the apparat
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, apparat_id):
return True if self.query_db("SELECT appnr FROM semesterapparat WHERE appnr=?", (apparat_id,), 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:
app_id (Union[str, int]): the id of the apparat
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
# Statistics
def statistic_request(self, **kwargs: Any):
"""Take n amount of kwargs and return the result of the query
"""
def __query(query):
"""execute the query and return the result
Args:
query (str): the constructed query
Returns:
list: the result of the query
"""
conn = self.connect()
cursor = conn.cursor()
result = cursor.execute(query).fetchall()
@@ -647,11 +852,23 @@ class Database:
# Admin data
def getUser(self):
"""Get a single user from the database"""
return self.query_db("SELECT * FROM user", one=True)
def getUsers(self):
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
Args:
user (str): username that tries to login
hashed_password (str): the password the user tries to login with
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]
if salt is None:
return False
@@ -662,30 +879,68 @@ class Database:
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
Args:
user (str): username
new_password (str): the hashed password
"""
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))
def getRole(self, user):
"""get the role of the user
Args:
user (str): username
Returns:
str: the name of the role
"""
return self.query_db("SELECT role FROM user WHERE username=?", (user,), one=True)[0]
def getRoles(self):
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):
def checkUsername(self, user)->bool:
"""a check to see if the username is already present in the database
Args:
user (str): the username
Returns:
bool: True if the username is present, False if not
"""
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 a user based on passed data.
"""create an user from the AdminCommands class.
Args:
----
- username (str): Username to be used
- password (str): the salted password
- role (str): Role of the user
- salt (str): a random salt for the user
user (str): the username of the user
password (str): a hashed password
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))
def deleteUser(self, user):
"""delete an unser
Args:
user (str): username of the user
"""
self.query_db("DELETE FROM user WHERE username=?", (user,))
def updateUser(self, username, data:dict[str, str]):
"""changge the data of a user
Args:
username (str): the username of the user
data (dict[str, str]): the data to be changed
"""
conn = self.connect()
cursor = conn.cursor()
query = "UPDATE user SET "
@@ -699,13 +954,33 @@ class Database:
cursor.execute(query, params)
conn.commit()
self.close_connection(conn)
def getFacultyMember(self, name:str):
def getFacultyMember(self, name:str)->tuple:
"""get a faculty member based on the name
Args:
name (str): the name to be searched for
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, oldlname, oldfname):
def updateFacultyMember(self, data:dict, oldlname:str, oldfname:str):
"""update the data of a faculty member
Args:
data (dict): a dictionary containing the data to be updated
oldlname (str): the old last name of the faculty member
oldfname (str): the old first name of the faculty member
"""
placeholders = ", ".join([f"{i}=:{i} " for i in data.keys()])
query = f"UPDATE prof SET {placeholders} WHERE lname = :oldlname AND fname = :oldfname"
data["oldlname"] = oldlname
data["oldfname"] = oldfname
self.query_db(query, data)
def getFacultyMembers(self):
"""get a list of all faculty members
Returns:
list[tuple]: a list of tuples containing the faculty members
"""
return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof")