Module database
+
+
++Expand source code +
+import datetime
+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 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.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
+config = OmegaConf.load("config.yaml")
+logger = MyLogger(__name__)
+
+
+class Database:
+ """
+ Initialize the database and create the tables if they do not exist.
+ """
+ def __init__(self, db_path: str = None):
+ """
+ Default constructor for the database class
+
+ Args:
+ db_path (str, optional): Optional Path for testing / specific purposes. Defaults to None.
+ """
+ if db_path is None:
+ self.db_path = config.database.path + config.database.name
+ self.db_path = self.db_path.replace("~", str(Path.home()))
+ else:
+ self.db_path = db_path
+ 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
+
+ Returns:
+ Union[List[Tuple], None]: _description_
+ """
+ try:
+ with sql.connect(self.db_path) as conn:
+ cursor = conn.cursor()
+ cursor.execute("SELECT * FROM sqlite_master WHERE type='table'")
+ return cursor.fetchall()
+ except sql.OperationalError:
+ return None
+ def connect(self)->sql.Connection:
+ """
+ Connect to the database
+
+ Returns:
+ 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
+
+ Args:
+ ----
+ - conn (sql.Connection): the connection to be closed
+ """
+ conn.close()
+ def create_tables(self):
+ """
+ Create the tables in the database
+ """
+ conn = self.connect()
+ cursor = conn.cursor()
+ cursor.execute(CREATE_TABLE_APPARAT)
+ cursor.execute(CREATE_TABLE_MESSAGES)
+ cursor.execute(CREATE_TABLE_MEDIA)
+ cursor.execute(CREATE_TABLE_APPKONTOS)
+ cursor.execute(CREATE_TABLE_FILES)
+ cursor.execute(CREATE_TABLE_PROF)
+ cursor.execute(CREATE_TABLE_USER)
+ 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
+
+ Args:
+ query (str): The query to be executed
+ params (Tuple): the parameters to be inserted into the database
+ """
+ conn = self.connect()
+ cursor = conn.cursor()
+ logger.log_info(f"Inserting {params} into database with query {query}")
+ 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.
+
+ Args:
+ query (str): The query to be executed
+ args (Tuple, optional): The arguments for the query. Defaults to ().
+ one (bool, optional): Return the first result only. Defaults to False.
+
+ Returns:
+ Union[Typle|List[Tuple]]: Returns the result of the query
+ """
+ conn = self.connect()
+ cursor = conn.cursor()
+ logger.log_info(f"Querying database with query {query}, args: {args}")
+ cursor.execute(query, args)
+ rv = cursor.fetchall()
+ conn.commit()
+ self.close_connection(conn)
+ 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]):
+ """
+ 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.
+
+ Args:
+ bookdata (BookData): The metadata of the book to be added
+ app_id (str): The apparat id where the book should be added to
+ prof_id (str): The id of the professor where the book should be added to.
+ """
+ conn = self.connect()
+ cursor = conn.cursor()
+ t_query = (
+ f"SELECT bookdata FROM media WHERE app_id={app_id} AND prof_id={prof_id}"
+ )
+ # print(t_query)
+ result = cursor.execute(t_query).fetchall()
+ result = [load_pickle(i[0]) for i in result]
+ if bookdata in result:
+ print("Bookdata already in database")
+ # check if the book was deleted in the apparat
+ query = (
+ "SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?"
+ )
+ params = (app_id, prof_id, dump_pickle(bookdata))
+ result = cursor.execute(query, params).fetchone()
+ if result[0] == 1:
+ print("Book was deleted, updating bookdata")
+ query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND bookdata=?"
+ params = (app_id, prof_id, dump_pickle(bookdata))
+ cursor.execute(query, params)
+ conn.commit()
+ return
+
+ query = (
+ "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)"
+ )
+ converted = dump_pickle(bookdata)
+ params = (converted, app_id, prof_id, 0)
+ 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:
+ """
+ Get a book id based on the signature of the book.
+
+ Args:
+ app_id (str): The apparat id the book should be associated with
+ prof_id (str): The professor id the book should be associated with
+ signature (str): The signature of the book
+
+ 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]
+ 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:
+ """
+ Get the book based on the signature of the book.
+
+ Args:
+ app_id (str): The apparat id the book should be associated with
+ prof_id (str): The professor id the book should be associated with
+ signature (str): The signature of the book
+
+ 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))
+ 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:
+ """
+ Get the last book id in the database
+
+ Returns:
+ 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]]:
+ """
+ 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:
+ - signature: The signature of the book
+ - title: The title of the book
+
+ Returns:
+ list[tuple[BookData, int]]: A list of tuples containing the wrapped Metadata and the id of the book
+ """
+ rdata = self.query_db("SELECT * FROM media WHERE deleted=0")
+ ic(rdata, len(rdata))
+ mode = 0
+ if len(data)== 1:
+ if "signature" in data.keys():
+ mode = 1
+ elif "title" in data.keys():
+ mode = 2
+ elif len(data) == 2:
+ mode = 3
+ else:
+ return None
+ ret = []
+ for book in rdata:
+ bookdata = load_pickle(book[1])
+ app_id = book[2]
+ prof_id = book[3]
+ if mode == 1:
+ if data["signature"] in bookdata.signature:
+ ret.append((bookdata,app_id,prof_id))
+ elif mode == 2:
+ if data["title"] in bookdata.title:
+ 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))
+ ic(ret)
+ return ret
+ def setAvailability(self, book_id:str, available:str):
+ """
+ Set the availability of a book in the database
+
+ Args:
+ 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:
+ """
+ Get the id of a book based on the metadata of the book
+
+ Args:
+ bookdata (BookData): The wrapped metadata of the book
+ app_id (str): The apparat id the book should be associated with
+ prof_id (str): The professor id the book should be associated with
+
+ 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)
+ return result[0]
+ def getBook(self,book_id:int)->BookData:
+ """
+ Get the book based on the id in the database
+
+ Args:
+ book_id (int): The id of the book
+
+ 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]]:
+ """
+ Get the Books based on the apparat id and the professor id
+
+ Args:
+ app_id (str): The ID of the apparat
+ prof_id (str): The ID of the professor
+ deleted (int, optional): The state of the book. Set to 1 to include deleted ones. Defaults to 0.
+
+ 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'})")
+ ret_result = []
+ for result_a in qdata:
+ data = {"id": int, "bookdata": BookData, "available": int}
+ data["id"] = result_a[0]
+ data["bookdata"] = load_pickle(result_a[1])
+ 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
+
+ Args:
+ 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))
+ def deleteBook(self, book_id):
+ """
+ Delete a book from the database
+
+ Args:
+ book_id (str): ID of the book
+ """
+ self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,))
+
+ # File Interactions
+ def getBlob(self, filename, app_id:Union[str,int]):
+ """
+ Get a blob from the database
+
+ Args:
+ filename (str): The name of the file
+ app_id (str): ID of the apparat
+
+ Returns:
+ 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]):
+ """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"]
+ filetyp = f["type"]
+ if path == "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:
+ """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()))
+ tempdir_path = Path(tempdir)
+ if not os.path.exists(tempdir_path):
+ os.mkdir(tempdir_path)
+ file = tempfile.NamedTemporaryFile(
+ delete=False, dir=tempdir_path, mode="wb", suffix=f".{filetype}"
+ )
+ file.write(blob)
+ print("file created")
+ return file.name
+ 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)->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: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"],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,))
+ ret = [
+ {
+ "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
+
+ 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: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: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)->tuple:
+ """get all the data of a professor based on the name
+
+ 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:Union[str,int], fields:List[str])->tuple:
+ """A customisable function to get specific data of a professor based on the id
+
+ Args:
+ prof_id (Union[str,int]): the id of the professor
+ fields (List[str]): a list of fields to be returned
+
+ Returns:
+ tuple: a tuple containing the requested data
+ """
+ query = "SELECT "
+ for field in fields:
+ query += f"{field},"
+ query = query[:-1]
+ 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()
+ prof_lname = prof_details["profname"].split(",")[0]
+ prof_lname = prof_lname.strip()
+ 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)
+ query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)"
+ self.insertInto(query=query, params=params)
+ 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)->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")
+ apparat = ApparatData()
+ apparat.appname = result[1]
+ apparat.appnr = result[4]
+ apparat.dauerapp = True if result[7] == 1 else False
+ prof_data = self.getProfData(self.getProfNameById(result[2]))
+ apparat.profname = self.getProfNameById(result[2])
+ apparat.prof_mail = prof_data[0]
+ apparat.prof_tel = prof_data[1]
+ apparat.prof_title = prof_data[2]
+ apparat.app_fach = result[3]
+ apparat.erstellsemester = result[5]
+ apparat.semester = result[8]
+ apparat.deleted = result[9]
+ apparat.apparat_adis_id = result[11]
+ apparat.prof_adis_id = result[12]
+ return apparat
+ 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 = [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):
+ """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,app_id))
+ else:
+ 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)->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:
+ raise AppPresentError(app_id)
+
+ self.createProf(apparat.get_prof_details())
+ prof_id = self.getProfId(apparat.profname)
+ ic(prof_id)
+ query = f"INSERT OR IGNORE INTO semesterapparat (appnr, name, erstellsemester, dauer, prof_id, fach,deletion_status,konto) VALUES ('{apparat.appnr}', '{apparat.appname}', '{apparat.semester}', '{apparat.dauerapp}', {prof_id}, '{apparat.app_fach}', '{0}', '{SEMAP_MEDIA_ACCOUNTS[apparat.appnr]}')"
+ logger.log_info(query)
+ self.query_db(query)
+ return self.getApparatId(apparat.appname)
+ 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[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()
+ c_tmp = []
+ for i in data:
+ c_tmp.append((i[0], self.getProfNameById(i[1])))
+ query = (
+ f"SELECT name,prof_id FROM semesterapparat WHERE deleted_date='{semester}'"
+ )
+ result = cursor.execute(query).fetchall()
+ d_tmp = []
+ for i in result:
+ d_tmp.append((i[0], self.getProfNameById(i[1])))
+ # group the apparats by prof
+ c_ret = {}
+ for i in c_tmp:
+ if i[1] not in c_ret.keys():
+ c_ret[i[1]] = [i[0]]
+ else:
+ c_ret[i[1]].append(i[0])
+ d_ret = {}
+ for i in d_tmp:
+ if i[1] not in d_ret.keys():
+ d_ret[i[1]] = [i[0]]
+ else:
+ 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]]:
+ """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()
+ created = []
+ deleted = []
+ for semester in semesters:
+ query = f"SELECT COUNT(*) FROM semesterapparat WHERE erstellsemester='{semester}'"
+ result = cursor.execute(query).fetchone()
+ created.append(result[0])
+ query = f"SELECT COUNT(*) FROM semesterapparat WHERE deletion_status=1 AND deleted_date='{semester}'"
+ result = cursor.execute(query).fetchone()
+ deleted.append(result[0])
+ # store data in a tuple
+ ret = []
+ e_tuple = ()
+ for sem in semesters:
+ e_tuple = (
+ sem,
+ created[semesters.index(sem)],
+ deleted[semesters.index(sem)],
+ )
+ ret.append(e_tuple)
+ self.close_connection(conn)
+ 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))
+ 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: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,
+ apparat_data.app_fach,
+ apparat_data.dauerapp,
+ self.getProfId(apparat_data.profname),
+ apparat_data.appnr,
+ )
+ self.query_db(query, params)
+ 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, 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()
+ for result_a in result:
+ orig_value = result_a
+ prof_name = self.getProfNameById(result_a[2])
+ # replace the prof_id with the prof_name
+ result_a = list(result_a)
+ result_a[2] = prof_name
+ result_a = tuple(result_a)
+ 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)
+ if "dauer" in kwargs.keys():
+ kwargs["dauer"] = kwargs["dauer"].replace("Ja", "1").replace("Nein", "0")
+ query = "SELECT * FROM semesterapparat WHERE "
+ for key, value in kwargs.items() if kwargs.items() is not None else {}:
+ print(key, value)
+ query += f"{key}='{value}' AND "
+ print(query)
+ # remove deletesemester part from normal query, as this will be added to the database upon deleting the apparat
+ if "deletesemester" in kwargs.keys():
+ query = query.replace(
+ f"deletesemester='{kwargs['deletesemester']}' AND ", ""
+ )
+ if "endsemester" in kwargs.keys():
+ if "erstellsemester" in kwargs.keys():
+ query = query.replace(f"endsemester='{kwargs['endsemester']}' AND ", "")
+ query = query.replace(
+ f"erstellsemester='{kwargs['erstellsemester']} AND ", "xyz"
+ )
+ else:
+ query = query.replace(
+ f"endsemester='{kwargs['endsemester']}' AND ", "xyz"
+ )
+ print("replaced")
+ query = query.replace(
+ "xyz",
+ f"(erstellsemester='{kwargs['endsemester']}' OR verlängerung_bis='{kwargs['endsemester']}') AND ",
+ )
+ # remove all x="" parts from the query where x is a key in kwargs
+ query = query[:-5]
+ print(query)
+ return __query(query)
+
+ # Admin data
+ def getUser(self):
+ """Get a single user from the database"""
+ return self.query_db("SELECT * FROM user", one=True)
+ 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
+ hashed_password = salt + hashed_password
+ 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
+
+ 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)->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:
+ """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 an user from the AdminCommands class.
+
+ Args:
+ 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 "
+ for key,value in data.items():
+ if key == "username":
+ continue
+ query += f"{key}='{value}',"
+ query = query[:-1]
+ query += " WHERE username=?"
+ params = (username,)
+ cursor.execute(query, params)
+ conn.commit()
+ self.close_connection(conn)
+ 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: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")Classes
+-
+
+class Database +(db_path: str = None) +
+-
++
Initialize the database and create the tables if they do not exist.
+Default constructor for the database class
+Args
+-
+
db_path:str, optional
+- Optional Path for testing / specific purposes. Defaults to None. +
+++Expand source code +
+
+class Database: + """ + Initialize the database and create the tables if they do not exist. + """ + def __init__(self, db_path: str = None): + """ + Default constructor for the database class + + Args: + db_path (str, optional): Optional Path for testing / specific purposes. Defaults to None. + """ + if db_path is None: + self.db_path = config.database.path + config.database.name + self.db_path = self.db_path.replace("~", str(Path.home())) + else: + self.db_path = db_path + 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 + + Returns: + Union[List[Tuple], None]: _description_ + """ + try: + with sql.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute("SELECT * FROM sqlite_master WHERE type='table'") + return cursor.fetchall() + except sql.OperationalError: + return None + def connect(self)->sql.Connection: + """ + Connect to the database + + Returns: + 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 + + Args: + ---- + - conn (sql.Connection): the connection to be closed + """ + conn.close() + def create_tables(self): + """ + Create the tables in the database + """ + conn = self.connect() + cursor = conn.cursor() + cursor.execute(CREATE_TABLE_APPARAT) + cursor.execute(CREATE_TABLE_MESSAGES) + cursor.execute(CREATE_TABLE_MEDIA) + cursor.execute(CREATE_TABLE_APPKONTOS) + cursor.execute(CREATE_TABLE_FILES) + cursor.execute(CREATE_TABLE_PROF) + cursor.execute(CREATE_TABLE_USER) + 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 + + Args: + query (str): The query to be executed + params (Tuple): the parameters to be inserted into the database + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Inserting {params} into database with query {query}") + 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. + + Args: + query (str): The query to be executed + args (Tuple, optional): The arguments for the query. Defaults to (). + one (bool, optional): Return the first result only. Defaults to False. + + Returns: + Union[Typle|List[Tuple]]: Returns the result of the query + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Querying database with query {query}, args: {args}") + cursor.execute(query, args) + rv = cursor.fetchall() + conn.commit() + self.close_connection(conn) + 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]): + """ + 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. + + Args: + bookdata (BookData): The metadata of the book to be added + app_id (str): The apparat id where the book should be added to + prof_id (str): The id of the professor where the book should be added to. + """ + conn = self.connect() + cursor = conn.cursor() + t_query = ( + f"SELECT bookdata FROM media WHERE app_id={app_id} AND prof_id={prof_id}" + ) + # print(t_query) + result = cursor.execute(t_query).fetchall() + result = [load_pickle(i[0]) for i in result] + if bookdata in result: + print("Bookdata already in database") + # check if the book was deleted in the apparat + query = ( + "SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?" + ) + params = (app_id, prof_id, dump_pickle(bookdata)) + result = cursor.execute(query, params).fetchone() + if result[0] == 1: + print("Book was deleted, updating bookdata") + query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND bookdata=?" + params = (app_id, prof_id, dump_pickle(bookdata)) + cursor.execute(query, params) + conn.commit() + return + + query = ( + "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" + ) + converted = dump_pickle(bookdata) + params = (converted, app_id, prof_id, 0) + 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: + """ + Get a book id based on the signature of the book. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + 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] + 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: + """ + Get the book based on the signature of the book. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + 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)) + 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: + """ + Get the last book id in the database + + Returns: + 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]]: + """ + 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: + - signature: The signature of the book + - title: The title of the book + + Returns: + list[tuple[BookData, int]]: A list of tuples containing the wrapped Metadata and the id of the book + """ + rdata = self.query_db("SELECT * FROM media WHERE deleted=0") + ic(rdata, len(rdata)) + mode = 0 + if len(data)== 1: + if "signature" in data.keys(): + mode = 1 + elif "title" in data.keys(): + mode = 2 + elif len(data) == 2: + mode = 3 + else: + return None + ret = [] + for book in rdata: + bookdata = load_pickle(book[1]) + app_id = book[2] + prof_id = book[3] + if mode == 1: + if data["signature"] in bookdata.signature: + ret.append((bookdata,app_id,prof_id)) + elif mode == 2: + if data["title"] in bookdata.title: + 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)) + ic(ret) + return ret + def setAvailability(self, book_id:str, available:str): + """ + Set the availability of a book in the database + + Args: + 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: + """ + Get the id of a book based on the metadata of the book + + Args: + bookdata (BookData): The wrapped metadata of the book + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + + 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) + return result[0] + def getBook(self,book_id:int)->BookData: + """ + Get the book based on the id in the database + + Args: + book_id (int): The id of the book + + 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]]: + """ + Get the Books based on the apparat id and the professor id + + Args: + app_id (str): The ID of the apparat + prof_id (str): The ID of the professor + deleted (int, optional): The state of the book. Set to 1 to include deleted ones. Defaults to 0. + + 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'})") + ret_result = [] + for result_a in qdata: + data = {"id": int, "bookdata": BookData, "available": int} + data["id"] = result_a[0] + data["bookdata"] = load_pickle(result_a[1]) + 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 + + Args: + 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)) + def deleteBook(self, book_id): + """ + Delete a book from the database + + Args: + book_id (str): ID of the book + """ + self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,)) + + # File Interactions + def getBlob(self, filename, app_id:Union[str,int]): + """ + Get a blob from the database + + Args: + filename (str): The name of the file + app_id (str): ID of the apparat + + Returns: + 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]): + """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"] + filetyp = f["type"] + if path == "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: + """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())) + tempdir_path = Path(tempdir) + if not os.path.exists(tempdir_path): + os.mkdir(tempdir_path) + file = tempfile.NamedTemporaryFile( + delete=False, dir=tempdir_path, mode="wb", suffix=f".{filetype}" + ) + file.write(blob) + print("file created") + return file.name + 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)->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: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"],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,)) + ret = [ + { + "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 + + 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: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: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)->tuple: + """get all the data of a professor based on the name + + 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:Union[str,int], fields:List[str])->tuple: + """A customisable function to get specific data of a professor based on the id + + Args: + prof_id (Union[str,int]): the id of the professor + fields (List[str]): a list of fields to be returned + + Returns: + tuple: a tuple containing the requested data + """ + query = "SELECT " + for field in fields: + query += f"{field}," + query = query[:-1] + 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() + prof_lname = prof_details["profname"].split(",")[0] + prof_lname = prof_lname.strip() + 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) + query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)" + self.insertInto(query=query, params=params) + 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)->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") + apparat = ApparatData() + apparat.appname = result[1] + apparat.appnr = result[4] + apparat.dauerapp = True if result[7] == 1 else False + prof_data = self.getProfData(self.getProfNameById(result[2])) + apparat.profname = self.getProfNameById(result[2]) + apparat.prof_mail = prof_data[0] + apparat.prof_tel = prof_data[1] + apparat.prof_title = prof_data[2] + apparat.app_fach = result[3] + apparat.erstellsemester = result[5] + apparat.semester = result[8] + apparat.deleted = result[9] + apparat.apparat_adis_id = result[11] + apparat.prof_adis_id = result[12] + return apparat + 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 = [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): + """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,app_id)) + else: + 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)->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: + raise AppPresentError(app_id) + + self.createProf(apparat.get_prof_details()) + prof_id = self.getProfId(apparat.profname) + ic(prof_id) + query = f"INSERT OR IGNORE INTO semesterapparat (appnr, name, erstellsemester, dauer, prof_id, fach,deletion_status,konto) VALUES ('{apparat.appnr}', '{apparat.appname}', '{apparat.semester}', '{apparat.dauerapp}', {prof_id}, '{apparat.app_fach}', '{0}', '{SEMAP_MEDIA_ACCOUNTS[apparat.appnr]}')" + logger.log_info(query) + self.query_db(query) + return self.getApparatId(apparat.appname) + 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[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() + c_tmp = [] + for i in data: + c_tmp.append((i[0], self.getProfNameById(i[1]))) + query = ( + f"SELECT name,prof_id FROM semesterapparat WHERE deleted_date='{semester}'" + ) + result = cursor.execute(query).fetchall() + d_tmp = [] + for i in result: + d_tmp.append((i[0], self.getProfNameById(i[1]))) + # group the apparats by prof + c_ret = {} + for i in c_tmp: + if i[1] not in c_ret.keys(): + c_ret[i[1]] = [i[0]] + else: + c_ret[i[1]].append(i[0]) + d_ret = {} + for i in d_tmp: + if i[1] not in d_ret.keys(): + d_ret[i[1]] = [i[0]] + else: + 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]]: + """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() + created = [] + deleted = [] + for semester in semesters: + query = f"SELECT COUNT(*) FROM semesterapparat WHERE erstellsemester='{semester}'" + result = cursor.execute(query).fetchone() + created.append(result[0]) + query = f"SELECT COUNT(*) FROM semesterapparat WHERE deletion_status=1 AND deleted_date='{semester}'" + result = cursor.execute(query).fetchone() + deleted.append(result[0]) + # store data in a tuple + ret = [] + e_tuple = () + for sem in semesters: + e_tuple = ( + sem, + created[semesters.index(sem)], + deleted[semesters.index(sem)], + ) + ret.append(e_tuple) + self.close_connection(conn) + 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)) + 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: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, + apparat_data.app_fach, + apparat_data.dauerapp, + self.getProfId(apparat_data.profname), + apparat_data.appnr, + ) + self.query_db(query, params) + 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, 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() + for result_a in result: + orig_value = result_a + prof_name = self.getProfNameById(result_a[2]) + # replace the prof_id with the prof_name + result_a = list(result_a) + result_a[2] = prof_name + result_a = tuple(result_a) + 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) + if "dauer" in kwargs.keys(): + kwargs["dauer"] = kwargs["dauer"].replace("Ja", "1").replace("Nein", "0") + query = "SELECT * FROM semesterapparat WHERE " + for key, value in kwargs.items() if kwargs.items() is not None else {}: + print(key, value) + query += f"{key}='{value}' AND " + print(query) + # remove deletesemester part from normal query, as this will be added to the database upon deleting the apparat + if "deletesemester" in kwargs.keys(): + query = query.replace( + f"deletesemester='{kwargs['deletesemester']}' AND ", "" + ) + if "endsemester" in kwargs.keys(): + if "erstellsemester" in kwargs.keys(): + query = query.replace(f"endsemester='{kwargs['endsemester']}' AND ", "") + query = query.replace( + f"erstellsemester='{kwargs['erstellsemester']} AND ", "xyz" + ) + else: + query = query.replace( + f"endsemester='{kwargs['endsemester']}' AND ", "xyz" + ) + print("replaced") + query = query.replace( + "xyz", + f"(erstellsemester='{kwargs['endsemester']}' OR verlängerung_bis='{kwargs['endsemester']}') AND ", + ) + # remove all x="" parts from the query where x is a key in kwargs + query = query[:-5] + print(query) + return __query(query) + + # Admin data + def getUser(self): + """Get a single user from the database""" + return self.query_db("SELECT * FROM user", one=True) + 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 + hashed_password = salt + hashed_password + 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 + + 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)->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: + """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 an user from the AdminCommands class. + + Args: + 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 " + for key,value in data.items(): + if key == "username": + continue + query += f"{key}='{value}'," + query = query[:-1] + query += " WHERE username=?" + params = (username,) + cursor.execute(query, params) + conn.commit() + self.close_connection(conn) + 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: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")Methods
+-
+
+def addBookToDatabase(self, bookdata: src.logic.dataclass.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.
+Args
+-
+
bookdata:BookData
+- The metadata of the book to be added +
app_id:str
+- The apparat id where the book should be added to +
prof_id:str
+- The id of the professor where the book should be added to. +
+++Expand source code +
+
+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. + + Args: + bookdata (BookData): The metadata of the book to be added + app_id (str): The apparat id where the book should be added to + prof_id (str): The id of the professor where the book should be added to. + """ + conn = self.connect() + cursor = conn.cursor() + t_query = ( + f"SELECT bookdata FROM media WHERE app_id={app_id} AND prof_id={prof_id}" + ) + # print(t_query) + result = cursor.execute(t_query).fetchall() + result = [load_pickle(i[0]) for i in result] + if bookdata in result: + print("Bookdata already in database") + # check if the book was deleted in the apparat + query = ( + "SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?" + ) + params = (app_id, prof_id, dump_pickle(bookdata)) + result = cursor.execute(query, params).fetchone() + if result[0] == 1: + print("Book was deleted, updating bookdata") + query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND bookdata=?" + params = (app_id, prof_id, dump_pickle(bookdata)) + cursor.execute(query, params) + conn.commit() + return + + query = ( + "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" + ) + converted = dump_pickle(bookdata) + params = (converted, app_id, prof_id, 0) + cursor.execute(query, params) + conn.commit() + self.close_connection(conn)
+ +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 +
+++Expand source code +
+
+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"],app_id))
+ +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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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, 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 +
+++Expand source code +
+
+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
+ +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 +
+++Expand source code +
+
+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 close_connection(self, conn: sqlite3.Connection) +
+-
++
closes the connection to the database
+Args:
+- conn (sql.Connection): the connection to be closed ++++Expand source code +
+
+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 connect(self) ‑> sqlite3.Connection +
+-
++
Connect to the database
+Returns
+-
+
sql.Connection
+- The active connection to the database +
+++Expand source code +
+
+def connect(self)->sql.Connection: + """ + Connect to the database + + Returns: + sql.Connection: The active connection to the database + """ + return sql.connect(self.db_path)
+ +def createApparat(self, apparat: src.logic.dataclass.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 +
+++Expand source code +
+
+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: + raise AppPresentError(app_id) + + self.createProf(apparat.get_prof_details()) + prof_id = self.getProfId(apparat.profname) + ic(prof_id) + query = f"INSERT OR IGNORE INTO semesterapparat (appnr, name, erstellsemester, dauer, prof_id, fach,deletion_status,konto) VALUES ('{apparat.appnr}', '{apparat.appname}', '{apparat.semester}', '{apparat.dauerapp}', {prof_id}, '{apparat.app_fach}', '{0}', '{SEMAP_MEDIA_ACCOUNTS[apparat.appnr]}')" + logger.log_info(query) + self.query_db(query) + return self.getApparatId(apparat.appname)
+ +def createProf(self, prof_details: dict) +
+-
++
Create a professor in the database
+Args
+-
+
prof_details:dict
+- a dictionary containing the details of the professor +
+++Expand source code +
+
+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() + prof_lname = prof_details["profname"].split(",")[0] + prof_lname = prof_lname.strip() + 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) + query = "INSERT OR IGNORE INTO prof (titel, fname, lname, mail, telnr, fullname) VALUES (?, ?, ?, ?, ?, ?)" + self.insertInto(query=query, params=params)
+ +def createUser(self, user, password, role, salt) +
+-
++
create an user from the AdminCommands class.
+Args
+-
+
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 +
+++Expand source code +
+
+def createUser(self, user, password, role, salt): + """create an user from the AdminCommands class. + + Args: + 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 create_tables(self) +
+-
++
Create the tables in the database
+++Expand source code +
+
+def create_tables(self): + """ + Create the tables in the database + """ + conn = self.connect() + cursor = conn.cursor() + cursor.execute(CREATE_TABLE_APPARAT) + cursor.execute(CREATE_TABLE_MESSAGES) + cursor.execute(CREATE_TABLE_MEDIA) + cursor.execute(CREATE_TABLE_APPKONTOS) + cursor.execute(CREATE_TABLE_FILES) + cursor.execute(CREATE_TABLE_PROF) + cursor.execute(CREATE_TABLE_USER) + cursor.execute(CREATE_TABLE_SUBJECTS) + conn.commit() + self.close_connection(conn)
+ +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 +
+++Expand source code +
+
+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 deleteBook(self, book_id) +
+-
++
Delete a book from the database
+Args
+-
+
book_id:str
+- ID of the book +
+++Expand source code +
+
+def deleteBook(self, book_id): + """ + Delete a book from the database + + Args: + book_id (str): ID of the book + """ + self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,))
+ +def deleteMessage(self, message_id) +
+-
++
Delete a message from the database
+Args
+-
+
message_id:str
+- the id of the message +
+++Expand source code +
+
+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,))
+ +def deleteUser(self, user) +
+-
++
delete an unser
+Args
+-
+
user:str
+- username of the user +
+++Expand source code +
+
+def deleteUser(self, user): + """delete an unser + + Args: + user (str): username of the user + """ + self.query_db("DELETE FROM user WHERE username=?", (user,))
+ +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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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() + created = [] + deleted = [] + for semester in semesters: + query = f"SELECT COUNT(*) FROM semesterapparat WHERE erstellsemester='{semester}'" + result = cursor.execute(query).fetchone() + created.append(result[0]) + query = f"SELECT COUNT(*) FROM semesterapparat WHERE deletion_status=1 AND deleted_date='{semester}'" + result = cursor.execute(query).fetchone() + deleted.append(result[0]) + # store data in a tuple + ret = [] + e_tuple = () + for sem in semesters: + e_tuple = ( + sem, + created[semesters.index(sem)], + deleted[semesters.index(sem)], + ) + ret.append(e_tuple) + self.close_connection(conn) + return ret
+ +def getApparatData(self, appnr, appname) ‑> src.logic.dataclass.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 +
+++Expand source code +
+
+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") + apparat = ApparatData() + apparat.appname = result[1] + apparat.appnr = result[4] + apparat.dauerapp = True if result[7] == 1 else False + prof_data = self.getProfData(self.getProfNameById(result[2])) + apparat.profname = self.getProfNameById(result[2]) + apparat.prof_mail = prof_data[0] + apparat.prof_tel = prof_data[1] + apparat.prof_title = prof_data[2] + apparat.app_fach = result[3] + apparat.erstellsemester = result[5] + apparat.semester = result[8] + apparat.deleted = result[9] + apparat.apparat_adis_id = result[11] + apparat.prof_adis_id = result[12] + return apparat
+ +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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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[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 +
+++Expand source code +
+
+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() + c_tmp = [] + for i in data: + c_tmp.append((i[0], self.getProfNameById(i[1]))) + query = ( + f"SELECT name,prof_id FROM semesterapparat WHERE deleted_date='{semester}'" + ) + result = cursor.execute(query).fetchall() + d_tmp = [] + for i in result: + d_tmp.append((i[0], self.getProfNameById(i[1]))) + # group the apparats by prof + c_ret = {} + for i in c_tmp: + if i[1] not in c_ret.keys(): + c_ret[i[1]] = [i[0]] + else: + c_ret[i[1]].append(i[0]) + d_ret = {} + for i in d_tmp: + if i[1] not in d_ret.keys(): + d_ret[i[1]] = [i[0]] + else: + d_ret[i[1]].append(i[0]) + self.close_connection(conn) + return {"created": c_ret, "deleted": d_ret}
+ +def getBlob(self, filename, app_id: Union[str, int]) +
+-
++
Get a blob from the database
+Args
+-
+
filename:str
+- The name of the file +
app_id:str
+- ID of the apparat +
Returns
+-
+
bytes
+- The file stored in +
+++Expand source code +
+
+def getBlob(self, filename, app_id:Union[str,int]): + """ + Get a blob from the database + + Args: + filename (str): The name of the file + app_id (str): ID of the apparat + + Returns: + 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 getBook(self, book_id: int) ‑> src.logic.dataclass.BookData +
+-
++
Get the book based on the id in the database
+Args
+-
+
book_id:int
+- The id of the book +
Returns
+-
+
BookData
+- The metadata of the book wrapped in a BookData object +
+++Expand source code +
+
+def getBook(self,book_id:int)->BookData: + """ + Get the book based on the id in the database + + Args: + book_id (int): The id of the book + + 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 getBookBasedOnSignature(self, app_id: Union[str, int], prof_id: Union[str, int], signature: str) ‑> src.logic.dataclass.BookData +
+-
++
Get the book based on the signature of the book.
+Args
+-
+
app_id:str
+- The apparat id the book should be associated with +
prof_id:str
+- The professor id the book should be associated with +
signature:str
+- The signature of the book +
Returns
+-
+
BookData
+- The total metadata of the book wrapped in a BookData object +
+++Expand source code +
+
+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. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + 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)) + books = [load_pickle(i[0]) for i in result] + book = [i for i in books if i.signature == signature][0] + return book
+ +def getBookId(self, bookdata: src.logic.dataclass.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
+Args
+-
+
bookdata:BookData
+- The wrapped metadata of the book +
app_id:str
+- The apparat id the book should be associated with +
prof_id:str
+- The professor id the book should be associated with +
Returns
+-
+
int
+- ID of the book +
+++Expand source code +
+
+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 + + Args: + bookdata (BookData): The wrapped metadata of the book + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + + 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) + return result[0]
+ +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.
+Args
+-
+
app_id:str
+- The apparat id the book should be associated with +
prof_id:str
+- The professor id the book should be associated with +
signature:str
+- The signature of the book +
Returns
+-
+
int
+- The id of the book +
+++Expand source code +
+
+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. + + Args: + app_id (str): The apparat id the book should be associated with + prof_id (str): The professor id the book should be associated with + signature (str): The signature of the book + + 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] + book = [i for i in books if i[0].signature == signature][0][1] + return book
+ +def getBooks(self, app_id: Union[str, int], prof_id: Union[str, int], deleted=0) ‑> list[dict[int, src.logic.dataclass.BookData, int]] +
+-
++
Get the Books based on the apparat id and the professor id
+Args
+-
+
app_id:str
+- The ID of the apparat +
prof_id:str
+- The ID of the professor +
deleted:int, optional
+- The state of the book. Set to 1 to include deleted ones. Defaults to 0. +
Returns
+-
+
list[dict[int, BookData, int]]
+- A list of dictionaries containing the id, the metadata of the book and the availability of the book +
+++Expand source code +
+
+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 + + Args: + app_id (str): The ID of the apparat + prof_id (str): The ID of the professor + deleted (int, optional): The state of the book. Set to 1 to include deleted ones. Defaults to 0. + + 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'})") + ret_result = [] + for result_a in qdata: + data = {"id": int, "bookdata": BookData, "available": int} + data["id"] = result_a[0] + data["bookdata"] = load_pickle(result_a[1]) + data["available"] = result_a[2] + ret_result.append(data) + return ret_result
+ +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 +
+++Expand source code +
+
+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 getFacultyMembers(self) +
+-
++
get a list of all faculty members
+Returns
+-
+
list[tuple]
+- a list of tuples containing the faculty members +
+++Expand source code +
+
+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")
+ +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 +
+++Expand source code +
+
+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 getLastBookId(self) ‑> int +
+-
++
Get the last book id in the database
+Returns
+-
+
int
+- ID of the last book in the database +
+++Expand source code +
+
+def getLastBookId(self)->int: + """ + Get the last book id in the database + + Returns: + 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 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 +
+++Expand source code +
+
+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,)) + ret = [ + { + "message": i[2], + "user": __get_user_name(i[4]), + "appnr": i[5], + "id": i[0] + } + for i in messages + ] + return ret
+ +def getProfByName(self, prof_name: str) ‑> tuple +
+-
++
get all the data of a professor based on the name
+Args
+-
+
prof_name:str
+- the name of the professor +
Returns
+-
+
tuple
+- the data of the professor +
+++Expand source code +
+
+def getProfByName(self, prof_name:str)->tuple: + """get all the data of a professor based on the name + + 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 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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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 getProfs(self) ‑> list[tuple] +
+-
++
Return all the professors in the database
+Returns
+-
+
list[tuple]
+- a list containing all the professors in individual tuples +
+++Expand source code +
+
+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")
+ +def getRole(self, user) +
+-
++
get the role of the user
+Args
+-
+
user:str
+- username +
Returns
+-
+
str
+- the name of the role +
+++Expand source code +
+
+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) ‑> list[tuple] +
+-
++
get all the roles in the database
+Returns
+-
+
list[str]
+- a list of all the roles +
+++Expand source code +
+
+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 getSemersters(self) ‑> list[str] +
+-
++
Return all the unique semesters in the database
+Returns
+-
+
list
+- a list of strings containing the semesters +
+++Expand source code +
+
+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 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:Union[str,int]
+- the id of the professor +
fields:List[str]
+- a list of fields to be returned +
Returns
+-
+
tuple
+- a tuple containing the requested data +
+++Expand source code +
+
+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 (Union[str,int]): the id of the professor + fields (List[str]): a list of fields to be returned + + Returns: + tuple: a tuple containing the requested data + """ + query = "SELECT " + for field in fields: + query += f"{field}," + query = query[:-1] + query += " FROM prof WHERE id=?" + return self.query_db(query, (prof_id,), one=True)[0]
+ +def getSubjects(self) +
+-
++
Get all the subjects in the database
+Returns
+-
+
list[tuple]
+- a list of tuples containing the subjects +
+++Expand source code +
+
+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")
+ +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 +
+++Expand source code +
+
+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 ""
+
+ -
++
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 +
+++Expand source code +
+
+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 = [i[0] for i in numbers] + logger.log_info(f"Currently used apparat numbers: {numbers}") + return numbers
+ +def getUser(self) +
+-
++
Get a single user from the database
+++Expand source code +
+
+def getUser(self): + """Get a single user from the database""" + return self.query_db("SELECT * FROM user", one=True)
+ +def getUsers(self) ‑> list[tuple] +
+-
++
Return a list of tuples of all the users in the database
+++Expand source code +
+
+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 get_db_contents(self) ‑> Optional[List[Tuple]] +
+-
++
Get the contents of the
+Returns
+-
+
Union[List[Tuple], None]
+- description +
+++Expand source code +
+
+def get_db_contents(self)->Union[List[Tuple], None]: + """ + Get the contents of the + + Returns: + Union[List[Tuple], None]: _description_ + """ + try: + with sql.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute("SELECT * FROM sqlite_master WHERE type='table'") + return cursor.fetchall() + except sql.OperationalError: + return None
+ +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 +
+++Expand source code +
+
+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"] + filetyp = f["type"] + if path == "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 insertInto(self, query: str, params: Tuple) ‑> None +
+-
++
Insert sent data into the database
+Args
+-
+
query:str
+- The query to be executed +
params:Tuple
+- the parameters to be inserted into the database +
+++Expand source code +
+
+def insertInto(self, query:str, params:Tuple) -> None: + """ + Insert sent data into the database + + Args: + query (str): The query to be executed + params (Tuple): the parameters to be inserted into the database + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Inserting {params} into database with query {query}") + cursor.execute(query, params) + conn.commit() + self.close_connection(conn)
+ +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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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 + hashed_password = salt + hashed_password + password = self.query_db("SELECT password FROM user WHERE username=?", (user,), one=True)[0] + if password == hashed_password: + return True + else: + return False
+ +def query_db(self, query: str, args: Tuple = (), one: bool = False) ‑> Union[Tuple, List[Tuple]] +
+-
++
Query the Database for the sent query.
+Args
+-
+
query:str
+- The query to be executed +
args:Tuple, optional
+- The arguments for the query. Defaults to (). +
one:bool, optional
+- Return the first result only. Defaults to False. +
Returns
+Union[Typle|List[Tuple]]: Returns the result of the query
+++Expand source code +
+
+def query_db(self, query: str, args: Tuple = (), one: bool = False)->Union[Tuple, List[Tuple]]: + """ + Query the Database for the sent query. + + Args: + query (str): The query to be executed + args (Tuple, optional): The arguments for the query. Defaults to (). + one (bool, optional): Return the first result only. Defaults to False. + + Returns: + Union[Typle|List[Tuple]]: Returns the result of the query + """ + conn = self.connect() + cursor = conn.cursor() + logger.log_info(f"Querying database with query {query}, args: {args}") + cursor.execute(query, args) + rv = cursor.fetchall() + conn.commit() + self.close_connection(conn) + return (rv[0] if rv else None) if one else rv
+ +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 +
+++Expand source code +
+
+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())) + tempdir_path = Path(tempdir) + if not os.path.exists(tempdir_path): + os.mkdir(tempdir_path) + file = tempfile.NamedTemporaryFile( + delete=False, dir=tempdir_path, mode="wb", suffix=f".{filetype}" + ) + file.write(blob) + print("file created") + return file.name
+ +def searchBook(self, data: dict[str, str]) ‑> list[tuple[src.logic.dataclass.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: +
-
+
- signature: The signature of the book +
- title: The title of the book +
Returns
+-
+
list[tuple[BookData, int]]
+- A list of tuples containing the wrapped Metadata and the id of the book +
+++Expand source code +
+
+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: + - signature: The signature of the book + - title: The title of the book + + Returns: + list[tuple[BookData, int]]: A list of tuples containing the wrapped Metadata and the id of the book + """ + rdata = self.query_db("SELECT * FROM media WHERE deleted=0") + ic(rdata, len(rdata)) + mode = 0 + if len(data)== 1: + if "signature" in data.keys(): + mode = 1 + elif "title" in data.keys(): + mode = 2 + elif len(data) == 2: + mode = 3 + else: + return None + ret = [] + for book in rdata: + bookdata = load_pickle(book[1]) + app_id = book[2] + prof_id = book[3] + if mode == 1: + if data["signature"] in bookdata.signature: + ret.append((bookdata,app_id,prof_id)) + elif mode == 2: + if data["title"] in bookdata.title: + 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)) + ic(ret) + return ret
+ +def setAvailability(self, book_id: str, available: str) +
+-
++
Set the availability of a book in the database
+Args
+-
+
book_id:str
+- The id of the book +
available:str
+- The availability of the book +
+++Expand source code +
+
+def setAvailability(self, book_id:str, available:str): + """ + Set the availability of a book in the database + + Args: + 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 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. +
+++Expand source code +
+
+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,app_id)) + else: + self.query_db("UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date,app_id))
+ +def statistic_request(self, **kwargs: Any) +
+-
++
Take n amount of kwargs and return the result of the query
+++Expand source code +
+
+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() + for result_a in result: + orig_value = result_a + prof_name = self.getProfNameById(result_a[2]) + # replace the prof_id with the prof_name + result_a = list(result_a) + result_a[2] = prof_name + result_a = tuple(result_a) + 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) + if "dauer" in kwargs.keys(): + kwargs["dauer"] = kwargs["dauer"].replace("Ja", "1").replace("Nein", "0") + query = "SELECT * FROM semesterapparat WHERE " + for key, value in kwargs.items() if kwargs.items() is not None else {}: + print(key, value) + query += f"{key}='{value}' AND " + print(query) + # remove deletesemester part from normal query, as this will be added to the database upon deleting the apparat + if "deletesemester" in kwargs.keys(): + query = query.replace( + f"deletesemester='{kwargs['deletesemester']}' AND ", "" + ) + if "endsemester" in kwargs.keys(): + if "erstellsemester" in kwargs.keys(): + query = query.replace(f"endsemester='{kwargs['endsemester']}' AND ", "") + query = query.replace( + f"erstellsemester='{kwargs['erstellsemester']} AND ", "xyz" + ) + else: + query = query.replace( + f"endsemester='{kwargs['endsemester']}' AND ", "xyz" + ) + print("replaced") + query = query.replace( + "xyz", + f"(erstellsemester='{kwargs['endsemester']}' OR verlängerung_bis='{kwargs['endsemester']}') AND ", + ) + # remove all x="" parts from the query where x is a key in kwargs + query = query[:-5] + print(query) + return __query(query)
+ +def updateApparat(self, apparat_data: src.logic.dataclass.ApparatData) +
+-
++
Update an apparat in the database
+Args
+-
+
apparat_data:ApparatData
+- the new metadata of the apparat +
+++Expand source code +
+
+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, + apparat_data.app_fach, + apparat_data.dauerapp, + self.getProfId(apparat_data.profname), + apparat_data.appnr, + ) + self.query_db(query, params)
+ +def updateBookdata(self, book_id, bookdata: src.logic.dataclass.BookData) +
+-
++
Update the bookdata in the database
+Args
+-
+
book_id:str
+- The id of the book +
bookdata:BookData
+- The new metadata of the book +
+++Expand source code +
+
+def updateBookdata(self, book_id, bookdata:BookData): + """ + Update the bookdata in the database + + Args: + 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))
+ +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 +
+++Expand source code +
+
+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 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 +
+++Expand source code +
+
+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 " + for key,value in data.items(): + if key == "username": + continue + query += f"{key}='{value}'," + query = query[:-1] + query += " WHERE username=?" + params = (username,) + cursor.execute(query, params) + conn.commit() + self.close_connection(conn)
+
+