from ..extensions import db from ..tools.encryption import decrypt, encrypt from ..tools.logs import write from flask import current_app as app from flask.helpers import flash from flask_login import current_user from sqlalchemy.exc import SQLAlchemyError from werkzeug.utils import secure_filename from datetime import datetime from json import dump from os import path, remove from pathlib import Path from uuid import uuid4 class Dataset(db.Model): id = db.Column(db.String(36), primary_key=True) name = db.Column(db.String(128), nullable=False) tests = db.relationship('Test', backref='dataset') creator_id = db.Column(db.String(36), db.ForeignKey('user.id')) date = db.Column(db.DateTime, nullable=False) default = db.Column(db.Boolean, default=False, nullable=True) accessed = db.Column(db.DateTime, nullable=True) locked = db.Column(db.Boolean, default=False, nullable=True) def __repr__(self): return f'.' @property def generate_id(self): raise AttributeError('generate_id is not a readable attribute.') generate_id.setter def generate_id(self): self.id = uuid4().hex @property def set_name(self): raise AttributeError('set_name is not a readable attribute.') set_name.setter def set_name(self, name:str): self.name = encrypt(name) def get_name(self): return decrypt(self.name) def make_default(self): try: for dataset in Dataset.query.all(): dataset.default = False except (SQLAlchemyError, ConnectionError) as exception: write('system.log', f'Database error when setting default dataset {self.id}: {exception}') return False, f'Database error {exception}.' self.default = True try: db.session.commit() except (SQLAlchemyError, ConnectionError) as exception: db.session.rollback() write('system.log', f'Database error when setting default dataset {self.id}: {exception}') return False, f'Database error {exception}.' write('system.log', f'Dataset {self.id} set as default by {current_user.get_username()}.') flash(message='Dataset set as default.', category='success') return True, f'Dataset set as default.' def delete(self): if self.default: message = 'Cannot delete the default dataset.' flash(message, 'error') return False, message try: if Dataset.query.count() == 1: message = 'Cannot delete the only dataset.' flash(message, 'error') return False, message except (SQLAlchemyError, ConnectionError) as exception: write('system.log', f'Database error when setting default dataset {self.id}: {exception}') return False, f'Database error {exception}.' write('system.log', f'Dataset {self.id} deleted by {current_user.get_username()}.') filename = secure_filename('.'.join([self.id,'json'])) data = Path(app.config.get('DATA')) file_path = path.join(data, 'questions', filename) try: db.session.delete(self) db.session.commit() except (SQLAlchemyError, ConnectionError) as exception: db.session.rollback() write('system.log', f'Database error when trying to delete dataset {self.id}: {exception}') return False, f'Database error: {exception}' remove(file_path) return True, 'Dataset deleted.' def create(self, data:list, default:bool=False): self.generate_id() timestamp = datetime.now() file_path = self.get_file() with open(file_path, 'w') as file: dump(data, file, indent=2) self.date = timestamp self.creator = current_user if default: self.make_default() write('system.log', f'New dataset {self.get_name()} added by {current_user.get_username()}.') try: db.session.add(self) db.session.commit() except (SQLAlchemyError, ConnectionError) as exception: db.session.rollback() write('system.log', f'Database error when trying to crreate dataset {self.id}: {exception}') return False, f'Database error: {exception}' return True, 'Dataset created.' def check_file(self): filename = secure_filename('.'.join([self.id,'json'])) data = Path(app.config.get('DATA')) file_path = path.join(data, 'questions', filename) if not path.isfile(file_path): return False, 'Data file is missing.' return True, 'Data file found.' def get_file(self): filename = secure_filename('.'.join([self.id,'json'])) data = Path(app.config.get('DATA')) file_path = path.join(data, 'questions', filename) return file_path def update(self, data:list=None, default:bool=False): self.date = datetime.now() if default: self.make_default() file_path = self.get_file() with open(file_path, 'w') as file: dump(data, file, indent=2) write('system.log', f'Dataset {self.id} edited by {current_user.get_username()}.') flash(f'Dataset {self.get_name()} successfully edited.', 'success') try: db.session.add(self) db.session.commit() except (SQLAlchemyError, ConnectionError) as exception: db.session.rollback() write('system.log', f'Database error when trying to update dataset {self.id}: {exception}') return False, f'Database error: {exception}' return True, 'Dataset successfully edited.'