94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
from datetime import datetime
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from tinydb import TinyDB, Query
|
|
import secrets
|
|
import pandas as pd
|
|
from pydantic import BaseModel, field_serializer
|
|
|
|
from .his_io import dtypes
|
|
|
|
|
|
pandas_json_format = 'split'
|
|
|
|
|
|
class Session(BaseModel):
|
|
token: str
|
|
table_name: str = ''
|
|
students_df: pd.DataFrame
|
|
points_df: pd.DataFrame
|
|
his_header: pd.DataFrame
|
|
grade_1_0: float = 0.92
|
|
grade_4_0: float = 0.4
|
|
max_points: int = 100
|
|
created: datetime = datetime.now()
|
|
edited: datetime = datetime.now()
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
return isinstance(other, Session) and \
|
|
self.token == other.token and \
|
|
self.table_name == other.table_name and \
|
|
self.students_df.equals(other.students_df) and \
|
|
self.points_df.equals(other.points_df) and \
|
|
self.his_header.equals(other.his_header) and \
|
|
self.grade_1_0 == other.grade_1_0 and \
|
|
self.grade_4_0 == other.grade_4_0 and \
|
|
self.max_points == other.max_points and \
|
|
self.created == other.created and \
|
|
self.edited == other.edited
|
|
|
|
@field_serializer('students_df', 'points_df')
|
|
def serialize_df(self, df: pd.DataFrame):
|
|
return df.reset_index().to_json(orient=pandas_json_format)
|
|
|
|
@field_serializer('his_header')
|
|
def serialize_his_header(self, df):
|
|
return df.to_json(orient=pandas_json_format)
|
|
|
|
@field_serializer('created', 'edited')
|
|
def serialize_datetime(self, dt):
|
|
return dt.isoformat()
|
|
|
|
class Config:
|
|
arbitrary_types_allowed = True
|
|
|
|
|
|
class DB:
|
|
def __init__(self, db_path='db.json'):
|
|
self.db_path = Path(db_path)
|
|
self.db = TinyDB(self.db_path, access_mode='r+')
|
|
|
|
def create_session(self):
|
|
token = secrets.token_urlsafe(16)
|
|
self.db.insert({'token': token})
|
|
|
|
return Session(**{
|
|
'token': token,
|
|
'students_df': pd.DataFrame(),
|
|
'points_df': pd.DataFrame(),
|
|
'his_header': pd.DataFrame(),
|
|
})
|
|
|
|
def get_session(self, token: str) -> Session:
|
|
results = self.db.search(Query().token == token)
|
|
|
|
if len(results) == 0:
|
|
raise ValueError(f'Session with token {token} not found')
|
|
|
|
session = results[0].copy()
|
|
session['points_df'] = pd.read_json(StringIO(session['points_df']), dtype=dtypes.students_table, orient=pandas_json_format).set_index('Matrikelnummer')
|
|
session['students_df'] = pd.read_json(StringIO(session['students_df']), dtype=dtypes.students_table, orient=pandas_json_format).set_index('Matrikelnummer')
|
|
session['his_header'] = pd.read_json(StringIO(session['his_header']), dtype=dtypes.his_header, orient=pandas_json_format)
|
|
return Session(**session)
|
|
|
|
def save_session(self, session: Session):
|
|
write_session = session.model_dump()
|
|
|
|
self.db.upsert(write_session, Query().token == write_session['token'])
|
|
|
|
def delete_session(self, session: Session):
|
|
self.db.remove(Query().token == session.token)
|
|
|
|
def list_tokens(self):
|
|
return [session['token'] for session in self.db.all()]
|