refactor docstring from custom format to google

This commit is contained in:
WorldTeacher
2024-02-14 15:49:02 +01:00
parent e4a8ac7d41
commit aa828f0afe

View File

@@ -3,7 +3,6 @@ import os
import re import re
import sqlite3 as sql import sqlite3 as sql
import tempfile import tempfile
import pickle
from src.logic.log import MyLogger from src.logic.log import MyLogger
from icecream import ic from icecream import ic
from typing import List, Tuple, Dict, Any, Optional, Union from typing import List, Tuple, Dict, Any, Optional, Union
@@ -19,7 +18,16 @@ logger = MyLogger(__name__)
class Database: class Database:
"""
Initialize the database and create the tables if they do not exist.
"""
def __init__(self, db_path: str = None): 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: if db_path is None:
self.db_path = config.database.path + config.database.name self.db_path = config.database.path + config.database.name
self.db_path = self.db_path.replace("~", str(Path.home())) 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") logger.log_critical("Database does not exist, creating tables")
self.create_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: try:
with sql.connect(self.db_path) as conn: with sql.connect(self.db_path) as conn:
cursor = conn.cursor() cursor = conn.cursor()
@@ -38,13 +52,29 @@ class Database:
except sql.OperationalError: except sql.OperationalError:
return None 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) return sql.connect(self.db_path)
def close_connection(self, conn: sql.Connection): def close_connection(self, conn: sql.Connection):
"""
closes the connection to the database
Args:
----
- conn (sql.Connection): the connection to be closed
"""
conn.close() conn.close()
def create_tables(self): def create_tables(self):
"""
Create the tables in the database
"""
conn = self.connect() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(CREATE_TABLE_APPARAT) cursor.execute(CREATE_TABLE_APPARAT)
@@ -58,7 +88,14 @@ class Database:
conn.commit() conn.commit()
self.close_connection(conn) 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() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
logger.log_info(f"Inserting {params} into database with query {query}") logger.log_info(f"Inserting {params} into database with query {query}")
@@ -66,7 +103,18 @@ class Database:
conn.commit() conn.commit()
self.close_connection(conn) 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() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
logger.log_info(f"Querying database with query {query}, args: {args}") 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 return (rv[0] if rv else None) if one else rv
# Books # 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() conn = self.connect()
cursor = conn.cursor() cursor = conn.cursor()
t_query = ( t_query = (
@@ -85,19 +141,19 @@ class Database:
) )
# print(t_query) # print(t_query)
result = cursor.execute(t_query).fetchall() 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: if bookdata in result:
print("Bookdata already in database") print("Bookdata already in database")
# check if the book was deleted in the apparat # check if the book was deleted in the apparat
query = ( query = (
"SELECT deleted FROM media WHERE app_id=? AND prof_id=? AND bookdata=?" "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() result = cursor.execute(query, params).fetchone()
if result[0] == 1: if result[0] == 1:
print("Book was deleted, updating bookdata") print("Book was deleted, updating bookdata")
query = "UPDATE media SET deleted=0 WHERE app_id=? AND prof_id=? AND 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) cursor.execute(query, params)
conn.commit() conn.commit()
return return
@@ -105,25 +161,63 @@ class Database:
query = ( query = (
"INSERT INTO media (bookdata, app_id, prof_id,deleted) VALUES (?, ?, ?,?)" "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) params = (converted, app_id, prof_id, 0)
cursor.execute(query, params) cursor.execute(query, params)
conn.commit() conn.commit()
self.close_connection(conn) self.close_connection(conn)
def getBookIdBasedOnSignature(self, app_id:str, prof_id:str,signature:str)->int: 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)) 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] 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] book = [i for i in books if i[0].signature == signature][0][1]
return book return book
def getBookBasedOnSignature(self, app_id:str, prof_id:str,signature:str)->BookData: 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)) 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] books = [load_pickle(i[0]) for i in result]
book = [i for i in books if i.signature == signature][0] book = [i for i in books if i.signature == signature][0]
return book return book
def getLastBookId(self)->int: 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] 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]]: 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") rdata = self.query_db("SELECT * FROM media WHERE deleted=0")
ic(rdata, len(rdata)) ic(rdata, len(rdata))
mode = 0 mode = 0
@@ -152,47 +246,92 @@ class Database:
ret.append((bookdata,app_id,prof_id)) ret.append((bookdata,app_id,prof_id))
ic(ret) ic(ret)
return ret return ret
def setAvailability(self, book_id, available): def setAvailability(self, book_id:str, available:str):
self.query_db("UPDATE media SET available=? WHERE id=?", (available,book_id)) """
def getBookId(self, bookdata:BookData, app_id, prof_id): Set the availability of a book in the database
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.
Args: Args:
---- book_id (str): The id of the book
- app_id (int): ID of the apparat available (str): The availability of the book
- 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. 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: Returns:
------- int: ID of the book
- 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: """
------- 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)
{"id": int, return result[0]
"bookdata": BookData, def getBook(self,book_id:int)->BookData:
"available": int} """
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'})") 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 = [] ret_result = []
for result_a in qdata: for result_a in qdata:
data = {"id": int, "bookdata": BookData, "available": int} data = {"id": int, "bookdata": BookData, "available": int}
data["id"] = result_a[0] data["id"] = result_a[0]
data["bookdata"] = pickle.loads(result_a[1]) data["bookdata"] = load_pickle(result_a[1])
data["available"] = result_a[2] data["available"] = result_a[2]
ret_result.append(data) ret_result.append(data)
return ret_result return ret_result
def updateBookdata(self, book_id, bookdata:BookData): 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)) self.query_db("UPDATE media SET bookdata=? WHERE id=?", (dump_pickle(bookdata),book_id))
def deleteBook(self, 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,)) self.query_db("UPDATE media SET deleted=1 WHERE id=?", (book_id,))
# File Interactions # File Interactions
def getBlob(self, filename, app_id): 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] 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): def insertFile(self, file: list[dict], app_id: int, prof_id):
for f in file: for f in file:
@@ -263,12 +402,42 @@ class Database:
def getProfByName(self, prof_name:str): def getProfByName(self, prof_name:str):
return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True) return self.query_db("SELECT * FROM prof WHERE fullname=?", (prof_name,), one=True)
def getProfId(self, prof_name:str): 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(",", "")) data = self.getProfByName(prof_name.replace(",", ""))
if data is None: if data is None:
return None return None
else: else:
return data[0] 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): def getProfData(self, profname:str):
data = self.query_db("SELECT mail, telnr, titel FROM prof WHERE fullname=?", (profname.replace(",",""),), one=True) data = self.query_db("SELECT mail, telnr, titel FROM prof WHERE fullname=?", (profname.replace(",",""),), one=True)
return data return data
def createProf(self, prof_details:dict): def createProf(self, prof_details:dict):