proper formatting

This commit is contained in:
WorldTeacher
2024-02-22 21:31:54 +01:00
parent 16430705cb
commit 67d782e1b3

View File

@@ -3,16 +3,28 @@ import os
import re import re
import sqlite3 as sql import sqlite3 as sql
import tempfile 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 pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union
from icecream import ic
from omegaconf import OmegaConf 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.constants import SEMAP_MEDIA_ACCOUNTS
from src.logic.dataclass import ApparatData, BookData from src.logic.dataclass import ApparatData, BookData
from src.errors import NoResultError, AppPresentError from src.logic.log import MyLogger
from src.utils import load_pickle, dump_pickle,create_blob from src.utils import create_blob, dump_pickle, load_pickle
config = OmegaConf.load("config.yaml") config = OmegaConf.load("config.yaml")
logger = MyLogger(__name__) logger = MyLogger(__name__)
@@ -21,7 +33,8 @@ class Database:
""" """
Initialize the database and create the tables if they do not exist. 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 Default constructor for the database class
@@ -36,7 +49,8 @@ class Database:
if self.get_db_contents() is None: if self.get_db_contents() is None:
logger.log_critical("Database does not exist, creating tables") logger.log_critical("Database does not exist, creating tables")
self.create_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
@@ -50,14 +64,17 @@ class Database:
return cursor.fetchall() return cursor.fetchall()
except sql.OperationalError: except sql.OperationalError:
return None return None
def connect(self)->sql.Connection:
def connect(self) -> sql.Connection:
""" """
Connect to the database Connect to the database
Returns: Returns:
sql.Connection: The active connection to the database sql.Connection: The active connection to the database
""" """
print(self.db_path)
return sql.connect(self.db_path) return sql.connect(self.db_path)
def close_connection(self, conn: sql.Connection): def close_connection(self, conn: sql.Connection):
""" """
closes the connection to the database closes the connection to the database
@@ -67,6 +84,7 @@ class Database:
- conn (sql.Connection): the connection to be closed - conn (sql.Connection): the connection to be closed
""" """
conn.close() conn.close()
def create_tables(self): def create_tables(self):
""" """
Create the tables in the database Create the tables in the database
@@ -83,7 +101,8 @@ class Database:
cursor.execute(CREATE_TABLE_SUBJECTS) cursor.execute(CREATE_TABLE_SUBJECTS)
conn.commit() conn.commit()
self.close_connection(conn) 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 Insert sent data into the database
@@ -97,7 +116,10 @@ class Database:
cursor.execute(query, params) cursor.execute(query, params)
conn.commit() conn.commit()
self.close_connection(conn) 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. Query the Database for the sent query.
@@ -119,7 +141,9 @@ class Database:
return (rv[0] if rv else None) if one else rv return (rv[0] if rv else None) if one else rv
# Books # 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. 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) cursor.execute(query, params)
conn.commit() conn.commit()
self.close_connection(conn) 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. Get a book id based on the signature of the book.
@@ -172,11 +199,17 @@ class Database:
Returns: Returns:
int: The id of the book 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)) result = self.query_db(
books = [(load_pickle(i[0]),i[1]) for i in result] "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] book = [i for i in books if i[0].signature == signature][0][1]
return book 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. Get the book based on the signature of the book.
@@ -188,11 +221,14 @@ class Database:
Returns: Returns:
BookData: The total metadata of the book wrapped in a BookData object 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] books = [load_pickle(i[0]) for i in result]
book = [i for i in books if i.signature == signature][0] book = [i for i in books if i.signature == signature][0]
return book return book
def getLastBookId(self)->int:
def getLastBookId(self) -> int:
""" """
Get the last book id in the database Get the last book id in the database
@@ -200,7 +236,8 @@ class Database:
int: ID of the last book in the 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] 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. Search a book in the database based on the sent data.
@@ -215,7 +252,7 @@ class Database:
rdata = self.query_db("SELECT * FROM media WHERE deleted=0") rdata = self.query_db("SELECT * FROM media WHERE deleted=0")
ic(rdata, len(rdata)) ic(rdata, len(rdata))
mode = 0 mode = 0
if len(data)== 1: if len(data) == 1:
if "signature" in data.keys(): if "signature" in data.keys():
mode = 1 mode = 1
elif "title" in data.keys(): elif "title" in data.keys():
@@ -231,16 +268,20 @@ class Database:
prof_id = book[3] prof_id = book[3]
if mode == 1: if mode == 1:
if data["signature"] in bookdata.signature: if data["signature"] in bookdata.signature:
ret.append((bookdata,app_id,prof_id)) ret.append((bookdata, app_id, prof_id))
elif mode == 2: elif mode == 2:
if data["title"] in bookdata.title: if data["title"] in bookdata.title:
ret.append((bookdata,app_id,prof_id)) ret.append((bookdata, app_id, prof_id))
elif mode == 3: elif mode == 3:
if data["signature"] in bookdata.signature and data["title"] in bookdata.title: if (
ret.append((bookdata,app_id,prof_id)) data["signature"] in bookdata.signature
and data["title"] in bookdata.title
):
ret.append((bookdata, app_id, prof_id))
ic(ret) ic(ret)
return 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 Set the availability of a book in the database
@@ -248,8 +289,11 @@ class Database:
book_id (str): The id of the book book_id (str): The id of the book
available (str): The availability of the book available (str): The availability of the book
""" """
self.query_db("UPDATE media SET available=? WHERE id=?", (available,book_id)) 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:
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 Get the id of a book based on the metadata of the book
@@ -261,9 +305,14 @@ class Database:
Returns: Returns:
int: ID of the book 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] 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 Get the book based on the id in the database
@@ -273,8 +322,15 @@ class Database:
Returns: Returns:
BookData: The metadata of the book wrapped in a BookData object 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]) return load_pickle(
def getBooks(self, app_id:Union[str,int], prof_id:Union[str,int], deleted=0)->list[dict[int, BookData, int]]: 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 Get the Books based on the apparat id and the professor id
@@ -286,7 +342,9 @@ class Database:
Returns: Returns:
list[dict[int, BookData, int]]: A list of dictionaries containing the id, the metadata of the book and the availability of the book 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 = [] ret_result = []
for result_a in qdata: for result_a in qdata:
data = {"id": int, "bookdata": BookData, "available": int} data = {"id": int, "bookdata": BookData, "available": int}
@@ -295,7 +353,8 @@ class Database:
data["available"] = result_a[2] data["available"] = result_a[2]
ret_result.append(data) ret_result.append(data)
return ret_result return ret_result
def updateBookdata(self, book_id, bookdata:BookData):
def updateBookdata(self, book_id, bookdata: BookData):
""" """
Update the bookdata in the database Update the bookdata in the database
@@ -303,7 +362,10 @@ class Database:
book_id (str): The id of the book book_id (str): The id of the book
bookdata (BookData): The new metadata 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): def deleteBook(self, book_id):
""" """
Delete a book from the database Delete a book from the database
@@ -314,7 +376,7 @@ class Database:
self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,)) self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,))
# File Interactions # 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 Get a blob from the database
@@ -325,8 +387,15 @@ class Database:
Returns: 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] return self.query_db(
def insertFile(self, file: list[dict], app_id:Union[str,int], prof_id:Union[str,int]): "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 """Instert a list of files into the database
Args: Args:
@@ -343,8 +412,11 @@ class Database:
continue continue
blob = create_blob(path) blob = create_blob(path)
query = "INSERT OR IGNORE INTO files (filename, fileblob, app_id, filetyp,prof_id) VALUES (?, ?, ?, ?,?)" 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)) self.query_db(query, (filename, blob, app_id, filetyp, prof_id))
def recreateFile(self, filename:str, app_id:Union[str,int],filetype:str)->str:
def recreateFile(
self, filename: str, app_id: Union[str, int], filetype: str
) -> str:
"""Recreate a file from the database """Recreate a file from the database
Args: Args:
@@ -367,7 +439,8 @@ class Database:
file.write(blob) file.write(blob)
print("file created") print("file created")
return file.name 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 """Get all the files associated with the apparat and the professor
Args: Args:
@@ -377,9 +450,12 @@ class Database:
Returns: Returns:
list[tuple]: a list of tuples containing the filename and the filetype for the corresponding apparat and professor 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 """Return all the unique semesters in the database
Returns: Returns:
@@ -397,7 +473,7 @@ class Database:
return self.query_db("SELECT * FROM subjects") return self.query_db("SELECT * FROM subjects")
# Messages # 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 """add a Message to the database
Args: Args:
@@ -405,11 +481,19 @@ class Database:
user (str): the user who added the message user (str): the user who added the message
app_id (Union[str,int]): the id of the apparat app_id (Union[str,int]): the id of the apparat
""" """
def __getUserId(user): 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) 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)) self.query_db(
def getMessages(self, date:str)->list[dict[str, str, str, str]]: "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 """Get all the messages for a specific date
Args: Args:
@@ -418,19 +502,19 @@ class Database:
Returns: 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 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): 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,)) messages = self.query_db("SELECT * FROM messages WHERE remind_at=?", (date,))
ret = [ 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 for i in messages
] ]
return ret return ret
def deleteMessage(self, message_id): def deleteMessage(self, message_id):
"""Delete a message from the database """Delete a message from the database
@@ -440,7 +524,7 @@ class Database:
self.query_db("DELETE FROM messages WHERE id=?", (message_id,)) self.query_db("DELETE FROM messages WHERE id=?", (message_id,))
# Prof data # 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 """Get a professor name based on the id
Args: Args:
@@ -450,12 +534,15 @@ class Database:
Returns: Returns:
str: The name of the professor 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: if add_title:
return f"{self.getTitleById(prof_id)}{prof[0]}" return f"{self.getTitleById(prof_id)}{prof[0]}"
else: else:
return prof[0] 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 """get the title of a professor based on the id
Args: Args:
@@ -464,9 +551,12 @@ class Database:
Returns: Returns:
str: the title of the professor, with an added whitespace at the end, if no title is present, an empty string is returned 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 "" 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 """get all the data of a professor based on the name
Args: Args:
@@ -475,8 +565,11 @@ class Database:
Returns: Returns:
tuple: the data of the professor tuple: the data of the professor
""" """
return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True) return self.query_db(
def getProfId(self, prof_name:str)->Optional[int]: "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 """Get the id of a professor based on the name
Args: Args:
@@ -491,7 +584,8 @@ class Database:
return None return None
else: else:
return data[0] 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 """A customisable function to get specific data of a professor based on the id
Args: Args:
@@ -507,7 +601,8 @@ class Database:
query = query[:-1] query = query[:-1]
query += " FROM prof WHERE id=?" query += " FROM prof WHERE id=?"
return self.query_db(query, (prof_id,), one=True)[0] 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 """Get mail, telephone number and title of a professor based on the name
Args: Args:
@@ -516,9 +611,14 @@ class Database:
Returns: Returns:
tuple: the mail, telephone number and title of the professor 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 return data
def createProf(self, prof_details:dict):
def createProf(self, prof_details: dict):
"""Create a professor in the database """Create a professor in the database
Args: Args:
@@ -532,10 +632,18 @@ class Database:
prof_fullname = prof_details["profname"].replace(",", "") prof_fullname = prof_details["profname"].replace(",", "")
prof_mail = prof_details["prof_mail"] prof_mail = prof_details["prof_mail"]
prof_tel = prof_details["prof_tel"] 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 (?, ?, ?, ?, ?, ?)" query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)"
self.insertInto(query=query, params=params) self.insertInto(query=query, params=params)
def getProfs(self)->list[tuple]:
def getProfs(self) -> list[tuple]:
"""Return all the professors in the database """Return all the professors in the database
Returns: Returns:
@@ -544,7 +652,7 @@ class Database:
return self.query_db("SELECT * FROM prof") return self.query_db("SELECT * FROM prof")
# Apparat # Apparat
def getAllAparats(self,deleted=0)->list[tuple]: def getAllAparats(self, deleted=0) -> list[tuple]:
"""Get all the apparats in the database """Get all the apparats in the database
Args: Args:
@@ -553,8 +661,11 @@ class Database:
Returns: Returns:
list[tuple]: a list of tuples containing the apparats list[tuple]: a list of tuples containing the apparats
""" """
return self.query_db("SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,)) return self.query_db(
def getApparatData(self, appnr, appname)->ApparatData: "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 """Get the Apparat data based on the apparat number and the name
Args: Args:
@@ -567,7 +678,11 @@ class Database:
Returns: Returns:
ApparatData: the appended data of the apparat wrapped in an ApparatData object 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: if result is None:
raise NoResultError("No result found") raise NoResultError("No result found")
apparat = ApparatData() apparat = ApparatData()
@@ -586,17 +701,21 @@ class Database:
apparat.apparat_adis_id = result[11] apparat.apparat_adis_id = result[11]
apparat.prof_adis_id = result[12] apparat.prof_adis_id = result[12]
return apparat 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 """Get a list of all the apparat numbers in the database that are currently in use
Returns: Returns:
List[int]: the list of used apparat numbers 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] numbers = [i[0] for i in numbers]
logger.log_info(f"Currently used apparat numbers: {numbers}") logger.log_info(f"Currently used apparat numbers: {numbers}")
return 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 """Set the new semester date for an apparat
Args: Args:
@@ -606,10 +725,16 @@ class Database:
""" """
date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d") date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d")
if dauerapp: 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: else:
self.query_db("UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date,app_id)) self.query_db(
def getApparatId(self, apparat_name)->Optional[int]: "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 """get the id of an apparat based on the name
Args: Args:
@@ -618,12 +743,15 @@ class Database:
Returns: Returns:
Optional[int]: the id of the apparat, if the apparat is not found, None is returned 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: if data is None:
return None return None
else: else:
return data[0] return data[0]
def createApparat(self, apparat:ApparatData)->int:
def createApparat(self, apparat: ApparatData) -> int:
"""create the apparat in the database """create the apparat in the database
Args: Args:
@@ -648,7 +776,8 @@ class Database:
logger.log_info(query) logger.log_info(query)
self.query_db(query) self.query_db(query)
return self.getApparatId(apparat.appname) 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 """Get all apparats based on the professor id
Args: Args:
@@ -657,8 +786,11 @@ class Database:
Returns: Returns:
list[tuple]: a list of tuples containing the apparats list[tuple]: a list of tuples containing the apparats
""" """
return self.query_db("SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,)) return self.query_db(
def getApparatsBySemester(self, semester:str)->dict[list]: "SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,)
)
def getApparatsBySemester(self, semester: str) -> dict[list]:
"""get all apparats based on the semester """get all apparats based on the semester
Args: Args:
@@ -667,7 +799,10 @@ class Database:
Returns: Returns:
dict[list]: a list off all created and deleted apparats for the selected semester 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() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
c_tmp = [] c_tmp = []
@@ -695,7 +830,8 @@ class Database:
d_ret[i[1]].append(i[0]) d_ret[i[1]].append(i[0])
self.close_connection(conn) self.close_connection(conn)
return {"created": c_ret, "deleted": d_ret} 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 """get a list of all apparats created and deleted by semester
Returns: Returns:
@@ -725,14 +861,19 @@ class Database:
ret.append(e_tuple) ret.append(e_tuple)
self.close_connection(conn) self.close_connection(conn)
return ret return ret
def deleteApparat(self, app_id:Union[str,int], semester:str):
def deleteApparat(self, app_id: Union[str, int], semester: str):
"""Delete an apparat from the database """Delete an apparat from the database
Args: Args:
app_id (Union[str, int]): the id of the apparat app_id (Union[str, int]): the id of the apparat
semester (str): the semester the apparat should be deleted from 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): def isEternal(self, id):
"""check if the apparat is eternal (dauerapparat) """check if the apparat is eternal (dauerapparat)
@@ -742,8 +883,11 @@ class Database:
Returns: Returns:
int: the state of the apparat int: the state of the apparat
""" """
return self.query_db("SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True) return self.query_db(
def getApparatName(self, app_id:Union[str,int], prof_id:Union[str,int]): "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 """get the name of the apparat based on the id
Args: Args:
@@ -753,8 +897,13 @@ class Database:
Returns: Returns:
str: the name of the apparat 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] return self.query_db(
def updateApparat(self, apparat_data:ApparatData): "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 """Update an apparat in the database
Args: Args:
@@ -769,7 +918,8 @@ class Database:
apparat_data.appnr, apparat_data.appnr,
) )
self.query_db(query, params) 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 """check if the apparat is already present in the database based on the name
Args: Args:
@@ -778,8 +928,17 @@ class Database:
Returns: Returns:
bool: True if the apparat is present, False if not 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 return (
def checkApparatExistsById(self, app_id:Union[str,int])->bool: 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 """a check to see if the apparat is already present in the database, based on the id
Args: Args:
@@ -788,11 +947,18 @@ class Database:
Returns: Returns:
bool: True if the apparat is present, False if not 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 # Statistics
def statistic_request(self, **kwargs: Any): 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): def __query(query):
"""execute the query and return the result """execute the query and return the result
@@ -815,6 +981,7 @@ class Database:
result[result.index(orig_value)] = result_a result[result.index(orig_value)] = result_a
self.close_connection(conn) self.close_connection(conn)
return result return result
if "deletable" in kwargs.keys(): 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']}')" 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) return __query(query)
@@ -854,7 +1021,8 @@ class Database:
def getUser(self): def getUser(self):
"""Get a single user from the database""" """Get a single user from the database"""
return self.query_db("SELECT * FROM user", one=True) 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 a list of tuples of all the users in the database"""
return self.query_db("SELECT * FROM user") return self.query_db("SELECT * FROM user")
@@ -869,15 +1037,20 @@ class Database:
Returns: Returns:
bool: True if the login was successful, False if not 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: if salt is None:
return False return False
hashed_password = salt + hashed_password 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: if password == hashed_password:
return True return True
else: else:
return False return False
def changePassword(self, user, new_password): def changePassword(self, user, new_password):
"""change the password of a user. """change the password of a user.
The password will be added with the salt and then committed to the database The password will be added with the salt and then committed to the database
@@ -886,9 +1059,14 @@ class Database:
user (str): username user (str): username
new_password (str): the hashed password 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 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): def getRole(self, user):
"""get the role of the user """get the role of the user
@@ -898,15 +1076,19 @@ class Database:
Returns: Returns:
str: the name of the role str: the name of the role
""" """
return self.query_db("SELECT role FROM user WHERE username=?", (user,), one=True)[0] return self.query_db(
def getRoles(self)->list[tuple]: "SELECT role FROM user WHERE username=?", (user,), one=True
)[0]
def getRoles(self) -> list[tuple]:
"""get all the roles in the database """get all the roles in the database
Returns: Returns:
list[str]: a list of all the roles list[str]: a list of all the roles
""" """
return self.query_db("SELECT role FROM user") 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 """a check to see if the username is already present in the database
Args: Args:
@@ -915,8 +1097,11 @@ class Database:
Returns: Returns:
bool: True if the username is present, False if not 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 return True if data is not None else False
def createUser(self, user, password, role, salt): def createUser(self, user, password, role, salt):
"""create an user from the AdminCommands class. """create an user from the AdminCommands class.
@@ -926,7 +1111,11 @@ class Database:
role (str): the role of the user role (str): the role of the user
salt (str): a salt for the password 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): def deleteUser(self, user):
"""delete an unser """delete an unser
@@ -934,7 +1123,8 @@ class Database:
user (str): username of the user user (str): username of the user
""" """
self.query_db("DELETE FROM user WHERE username=?", (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 """changge the data of a user
Args: Args:
@@ -944,7 +1134,7 @@ class Database:
conn = self.connect() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
query = "UPDATE user SET " query = "UPDATE user SET "
for key,value in data.items(): for key, value in data.items():
if key == "username": if key == "username":
continue continue
query += f"{key}='{value}'," query += f"{key}='{value}',"
@@ -954,7 +1144,8 @@ class Database:
cursor.execute(query, params) cursor.execute(query, params)
conn.commit() conn.commit()
self.close_connection(conn) self.close_connection(conn)
def getFacultyMember(self, name:str)->tuple:
def getFacultyMember(self, name: str) -> tuple:
"""get a faculty member based on the name """get a faculty member based on the name
Args: Args:
@@ -963,8 +1154,13 @@ class Database:
Returns: Returns:
tuple: a tuple containing the data of the faculty member 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) return self.query_db(
def updateFacultyMember(self, data:dict, oldlname:str, oldfname:str): "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 """update the data of a faculty member
Args: Args:
@@ -977,6 +1173,7 @@ class Database:
data["oldlname"] = oldlname data["oldlname"] = oldlname
data["oldfname"] = oldfname data["oldfname"] = oldfname
self.query_db(query, data) self.query_db(query, data)
def getFacultyMembers(self): def getFacultyMembers(self):
"""get a list of all faculty members """get a list of all faculty members