From aa828f0afeb02e9ea6376b2735750fe467d1132d Mon Sep 17 00:00:00 2001 From: WorldTeacher <41587052+WorldTeacher@users.noreply.github.com> Date: Wed, 14 Feb 2024 15:49:02 +0100 Subject: [PATCH] refactor docstring from custom format to google --- src/backend/database.py | 231 ++++++++++++++++++++++++++++++++++------ 1 file changed, 200 insertions(+), 31 deletions(-) diff --git a/src/backend/database.py b/src/backend/database.py index 35028f4..214c115 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -3,7 +3,6 @@ import os import re import sqlite3 as sql import tempfile -import pickle from src.logic.log import MyLogger from icecream import ic from typing import List, Tuple, Dict, Any, Optional, Union @@ -19,7 +18,16 @@ logger = MyLogger(__name__) class Database: + """ + Initialize the database and create the tables if they do not exist. + """ def __init__(self, db_path: str = None): + """ + Default constructor for the database class + + Args: + db_path (str, optional): Optional Path for testing / specific purposes. Defaults to None. + """ if db_path is None: self.db_path = config.database.path + config.database.name self.db_path = self.db_path.replace("~", str(Path.home())) @@ -29,7 +37,13 @@ class Database: logger.log_critical("Database does not exist, creating tables") self.create_tables() - def get_db_contents(self): + 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() @@ -38,13 +52,29 @@ class Database: except sql.OperationalError: return None - def connect(self): + 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) @@ -58,7 +88,14 @@ class Database: conn.commit() self.close_connection(conn) - def insertInto(self, query, params): + 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}") @@ -66,7 +103,18 @@ class Database: conn.commit() self.close_connection(conn) - def query_db(self, query: str, args: Tuple = (), one: bool = 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 + """ conn = self.connect() cursor = conn.cursor() logger.log_info(f"Querying database with query {query}, args: {args}") @@ -77,7 +125,15 @@ class Database: return (rv[0] if rv else None) if one else rv # Books - def addBookToDatabase(self, bookdata:BookData,app_id:str, prof_id=str): + 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 = ( @@ -85,19 +141,19 @@ class Database: ) # print(t_query) result = cursor.execute(t_query).fetchall() - result = [pickle.loads(i[0]) for i in result] + 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, pickle.dumps(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, pickle.dumps(bookdata)) + params = (app_id, prof_id, dump_pickle(bookdata)) cursor.execute(query, params) conn.commit() return @@ -105,25 +161,63 @@ class Database: query = ( "INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" ) - converted = pickle.dumps(bookdata) + 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] - # ic(books) 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 @@ -152,47 +246,92 @@ class Database: ret.append((bookdata,app_id,prof_id)) ic(ret) return ret - def setAvailability(self, book_id, available): - self.query_db("UPDATE media SET available=? WHERE id=?", (available,book_id)) - def getBookId(self, bookdata:BookData, app_id, prof_id): - 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): - 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): - """request media from database and return result as list. + def setAvailability(self, book_id:str, available:str): + """ + Set the availability of a book in the database Args: - ---- - - app_id (int): ID of the apparat - - prof_id (int): ID of the prof - - del_state (int, optional): If deleted books should be requested as well. 1 = yes 0 = no. Defaults to 0. + 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: - ------- - - list[dict[int,BookData,int]]: Returns a list of dictionaries containing the bookdata, the id and the availability of the book in the following format: - ------- - {"id": int, - "bookdata": BookData, - "available": int} + 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"] = pickle.loads(result_a[1]) + 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: @@ -263,12 +402,42 @@ class Database: 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):