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:str, prof_id:str):
        """
        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:str, prof_id:str,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:str, prof_id:str,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, prof_id)->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, prof_id, 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):
        """
        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: int, prof_id):
        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, app_id,filetype):
        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:int, prof_id:int)->list[tuple]:
        return self.query_db("SELECT filename, filetyp FROM files WHERE app_id=? AND prof_id=?", (app_id,prof_id))

    def getSemersters(self):
        data = self.query_db("SELECT DISTINCT erstellsemester FROM semesterapparat")
        return [i[0] for i in data]

    def getSubjects(self):
        return self.query_db("SELECT * FROM subjects")

    # Messages
    def addMessage(self, message:dict,user, appnr):
        def __getUserId(user):
            return self.query_db("SELECT id FROM user WHERE username=?", (user,), one=True)[0]
        user_id = __getUserId(user)
        self.query_db("INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", (message["message"],user_id,message["remind_at"],appnr))
    def getMessages(self, date:str):
        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):
        self.query_db("DELETE FROM messages WHERE id=?", (message_id,))

    # Prof data
    def getProfNameById(self, prof_id:int,add_title:bool=False):
        prof = self.query_db("SELECT fullname FROM prof WHERE id=?", (prof_id,), one=True)
        if add_title:
            return f"{self.getTitleById(prof_id)}{prof[0]}"
        else:
            return prof[0]
    def getTitleById(self, prof_id:int):
        title = self.query_db("SELECT titel FROM prof WHERE id=?", (prof_id,), one=True)[0]
        return f"{title} " if title is not None else ""
    def getProfByName(self, prof_name:str):
        return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True)
    def getProfId(self, prof_name:str):
        """
        getProfId _summary_

        :param prof_name: _description_
        :type prof_name: str
        :return: _description_
        :rtype: _type_
        """
        data = self.getProfByName(prof_name.replace(",", ""))
        if data is None:
            return None
        else:
            return data[0]
    def getSpecificProfData(self, prof_id:int, fields:List[str]):
        """
        getSpecificProfData _summary_
        
        
        
        Args:
        ----
            - prof_id (int): _description_
            - fields (List[str]): _description_
        
        Returns:
        -------
            - _type_: _description_
        """
        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):
        
        data = self.query_db("SELECT mail, telnr, titel FROM prof WHERE fullname=?", (profname.replace(",",""),), one=True)
        return data
    def createProf(self, prof_details:dict):
        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):
        return self.query_db("SELECT * FROM prof")

    # Apparat
    def getAllAparats(self,deleted=0):
        return self.query_db("SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,))
    def getApparatData(self, appnr, appname)->ApparatData:

        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]:
        """
        getUnavailableApparatNumbers returns a list of all currently used ApparatNumbers
        
        
        
        Returns:
        -------
            - number(List[int]): a list of all currently used apparat numbers
        """
        numbers = self.query_db("SELECT appnr FROM semesterapparat WHERE deletion_status=0")    
        numbers = [i[0] for i in numbers]
        logger.log_info(f"Currently used apparat numbers: {numbers}")
        return numbers
    def setNewSemesterDate(self, appnr, newDate, dauerapp=False):
        date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d")
        if dauerapp:
            self.query_db("UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", (date,dauerapp,appnr))
        else:
            self.query_db("UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date,appnr))
    def getApparatId(self, apparat_name):
        data = self.query_db("SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True)
        if data is None:
            return None
        else:
            return data[0]
    def createApparat(self, apparat:ApparatData)->Optional[AppPresentError]|int:
        prof_id = self.getProfId(apparat.profname)
        app_id = self.getApparatId(apparat.appname)
        if app_id:
            return 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:int)->list[tuple]:
        return self.query_db("SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,))
    def getApparatsBySemester(self, semester:str)->dict:
        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]]:
        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, appnr, semester):
        self.query_db("UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", (semester,appnr)) 
    def isEternal(self, id):
        return self.query_db("SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True)
    def getApparatName(self, app_id, prof_id):
        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):
        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):
        return True if self.query_db("SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True) else False
    def checkApparatExistsById(self, apparat_id):
        return True if self.query_db("SELECT appnr FROM semesterapparat WHERE appnr=?", (apparat_id,), one=True) else False
    # Statistics
    def statistic_request(self, **kwargs: Any):
        def __query(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):
        return self.query_db("SELECT * FROM user", one=True)
    def getUsers(self):
        return self.query_db("SELECT * FROM user")

    def login(self, user, hashed_password):
        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):
        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):
        return self.query_db("SELECT role FROM user WHERE username=?", (user,), one=True)[0]
    def getRoles(self):
        return self.query_db("SELECT role FROM user")
    def checkUsername(self, user):
        data = self.query_db("SELECT username FROM user WHERE username=?", (user,), one=True)
        return True if data is not None else False
    def createUser(self, user, password, role, salt):
        """Create a user based on passed data.

        Args:
        ----
            - username (str): Username to be used
            - password (str): the salted password
            - role (str): Role of the user
            - salt (str): a random salt for the user
        """
        self.query_db("INSERT OR IGNORE INTO user (username, password, role, salt) VALUES (?,?,?,?)", (user,password,role,salt))
    def deleteUser(self, user):
        self.query_db("DELETE FROM user WHERE username=?", (user,))
    def updateUser(self, username, data:dict[str, str]):
        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):
        return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", (name,), one=True)
    def updateFacultyMember(self, data, oldlname, oldfname):
        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):
        return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof")

Methods

def addBookToDatabase(self, bookdata: src.logic.dataclass.BookData, app_id: str, prof_id: str)

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:str, prof_id:str):
    """
    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, appnr)

Expand source code

def addMessage(self, message:dict,user, appnr):
    def __getUserId(user):
        return self.query_db("SELECT id FROM user WHERE username=?", (user,), one=True)[0]
    user_id = __getUserId(user)
    self.query_db("INSERT INTO messages (message, user_id, remind_at,appnr) VALUES (?,?,?,?)", (message["message"],user_id,message["remind_at"],appnr))

def changePassword(self, user, new_password)

Expand source code

def changePassword(self, user, new_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)

Expand source code

def checkApparatExists(self, apparat_name):
    return True if self.query_db("SELECT appnr FROM semesterapparat WHERE name=?", (apparat_name,), one=True) else False

def checkApparatExistsById(self, apparat_id)

Expand source code

def checkApparatExistsById(self, apparat_id):
    return True if self.query_db("SELECT appnr FROM semesterapparat WHERE appnr=?", (apparat_id,), one=True) else False

def checkUsername(self, user)

Expand source code

def checkUsername(self, user):
    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) ‑> Union[src.errors.DatabaseErrors.AppPresentError, ForwardRef(None), int]

Expand source code

def createApparat(self, apparat:ApparatData)->Optional[AppPresentError]|int:
    prof_id = self.getProfId(apparat.profname)
    app_id = self.getApparatId(apparat.appname)
    if app_id:
        return 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)

Expand source code

def createProf(self, prof_details:dict):
    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 a user based on passed data.

Args:

- username (str): Username to be used
- password (str): the salted password
- role (str): Role of the user
- salt (str): a random salt for the user

Expand source code

def createUser(self, user, password, role, salt):
    """Create a user based on passed data.

    Args:
    ----
        - username (str): Username to be used
        - password (str): the salted password
        - role (str): Role of the user
        - salt (str): a random salt for the user
    """
    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, appnr, semester)

Expand source code

def deleteApparat(self, appnr, semester):
    self.query_db("UPDATE semesterapparat SET deletion_status=1, deleted_date=? WHERE appnr=?", (semester,appnr)) 

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)

Expand source code

def deleteMessage(self, message_id):
    self.query_db("DELETE FROM messages WHERE id=?", (message_id,))

def deleteUser(self, user)

Expand source code

def deleteUser(self, user):
    self.query_db("DELETE FROM user WHERE username=?", (user,))

def getAllAparats(self, deleted=0)

Expand source code

def getAllAparats(self,deleted=0):
    return self.query_db("SELECT * FROM semesterapparat WHERE deletion_status=?", (deleted,))

def getApparatCountBySemester(self) ‑> tuple[list[str], list[int]]

Expand source code

def getApparatCountBySemester(self)->tuple[list[str],list[int]]:
    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

Expand source code

def getApparatData(self, appnr, appname)->ApparatData:

    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)

Expand source code

def getApparatId(self, apparat_name):
    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, prof_id)

Expand source code

def getApparatName(self, app_id, prof_id):
    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: int) ‑> list[tuple]

Expand source code

def getApparatsByProf(self, prof_id:int)->list[tuple]:
    return self.query_db("SELECT * FROM semesterapparat WHERE prof_id=?", (prof_id,))

def getApparatsBySemester(self, semester: str) ‑> dict

Expand source code

def getApparatsBySemester(self, semester:str)->dict:
    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)

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):
    """
    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: str, prof_id: str, 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:str, prof_id:str,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, prof_id) ‑> 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, prof_id)->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: str, prof_id: str, 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:str, prof_id:str,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, prof_id, 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, prof_id, 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)

Expand source code

def getFacultyMember(self, name:str):
    return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof WHERE fullname=?", (name,), one=True)

def getFacultyMembers(self)

Expand source code

def getFacultyMembers(self):
    return self.query_db("SELECT titel, fname,lname,mail,telnr,fullname FROM prof")

def getFiles(self, app_id: int, prof_id: int) ‑> list[tuple]

Expand source code

def getFiles(self, app_id:int, prof_id:int)->list[tuple]:
    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)

Expand source code

def getMessages(self, date:str):
    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)

Expand source code

def getProfByName(self, prof_name:str):
    return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True)

def getProfData(self, profname: str)

Expand source code

def getProfData(self, profname:str):
    
    data = self.query_db("SELECT mail, telnr, titel FROM prof WHERE fullname=?", (profname.replace(",",""),), one=True)
    return data

def getProfId(self, prof_name: str)

getProfId summary

:param prof_name: description :type prof_name: str :return: description :rtype: type

Expand source code

def getProfId(self, prof_name:str):
    """
    getProfId _summary_

    :param prof_name: _description_
    :type prof_name: str
    :return: _description_
    :rtype: _type_
    """
    data = self.getProfByName(prof_name.replace(",", ""))
    if data is None:
        return None
    else:
        return data[0]

def getProfNameById(self, prof_id: int, add_title: bool = False)

Expand source code

def getProfNameById(self, prof_id:int,add_title:bool=False):
    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)

Expand source code

def getProfs(self):
    return self.query_db("SELECT * FROM prof")

def getRole(self, user)

Expand source code

def getRole(self, user):
    return self.query_db("SELECT role FROM user WHERE username=?", (user,), one=True)[0]

def getRoles(self)

Expand source code

def getRoles(self):
    return self.query_db("SELECT role FROM user")

def getSemersters(self)

Expand source code

def getSemersters(self):
    data = self.query_db("SELECT DISTINCT erstellsemester FROM semesterapparat")
    return [i[0] for i in data]

def getSpecificProfData(self, prof_id: int, fields: List[str])

getSpecificProfData summary

Args:

- prof_id (int): _description_
- fields (List[str]): _description_

Returns:

- _type_: _description_

Expand source code

def getSpecificProfData(self, prof_id:int, fields:List[str]):
    """
    getSpecificProfData _summary_
    
    
    
    Args:
    ----
        - prof_id (int): _description_
        - fields (List[str]): _description_
    
    Returns:
    -------
        - _type_: _description_
    """
    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)

Expand source code

def getSubjects(self):
    return self.query_db("SELECT * FROM subjects")

def getTitleById(self, prof_id: int)

Expand source code

def getTitleById(self, prof_id:int):
    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]

getUnavailableApparatNumbers returns a list of all currently used ApparatNumbers

Returns:

- number(List[int]): a list of all currently used apparat numbers

Expand source code

def getUnavailableApparatNumbers(self)->List[int]:
    """
    getUnavailableApparatNumbers returns a list of all currently used ApparatNumbers
    
    
    
    Returns:
    -------
        - number(List[int]): a list of all currently 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)

Expand source code

def getUser(self):
    return self.query_db("SELECT * FROM user", one=True)

def getUsers(self)

Expand source code

def getUsers(self):
    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: int, prof_id)

Expand source code

def insertFile(self, file: list[dict], app_id: int, prof_id):
    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)

Expand source code

def isEternal(self, id):
    return self.query_db("SELECT dauer FROM semesterapparat WHERE appnr=?", (id,), one=True)

def login(self, user, hashed_password)

Expand source code

def login(self, user, hashed_password):
    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, app_id, filetype)

Expand source code

def recreateFile(self, filename, app_id,filetype):
    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, appnr, newDate, dauerapp=False)

Expand source code

def setNewSemesterDate(self, appnr, newDate, dauerapp=False):
    date = datetime.datetime.strptime(newDate, "%d.%m.%Y").strftime("%Y-%m-%d")
    if dauerapp:
        self.query_db("UPDATE semesterapparat SET verlängerung_bis=?, dauerapp=? WHERE appnr=?", (date,dauerapp,appnr))
    else:
        self.query_db("UPDATE semesterapparat SET endsemester=? WHERE appnr=?", (date,appnr))

def statistic_request(self, **kwargs: Any)

Expand source code

def statistic_request(self, **kwargs: Any):
    def __query(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)

Expand source code

def updateApparat(self, apparat_data:ApparatData):
    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, oldlname, oldfname)

Expand source code

def updateFacultyMember(self, data, oldlname, oldfname):
    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])

Expand source code

def updateUser(self, username, data:dict[str, str]):
    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)