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)
-
-