Module database

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

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

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)