""" See the file DB.drawio for the design of the database. It can be opened in the internet browser with the website: https://app.diagrams.net For more information about the implementation, see the part to Models in the documentation of Flask-SQLAlchemy: https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/ """ from decimal import ROUND_HALF_UP, Decimal from flask import flash from flask_login import current_user from flask_security.models.fsqla_v3 import FsModels, FsRoleMixin, FsUserMixin from flask_sqlalchemy import SQLAlchemy from sqlalchemy import func, select from .exceptions import DatabaseException MIN_MARK = 0 MAX_MARK = 15 MIN_YEAR = 22 MAX_YEAR = 99 MIN_STUDENT_NUMBER = 0 MIN_EXPERIMENT_NUMBER = 1 MIN_GROUP_NUMBER = 1 MIN_DURATION_IN_DAYS = 1 MIN_PART_NUMBER = 1 db = SQLAlchemy() # For Flask-Security-Too FsModels.set_db_info(db) def get_first(table): return db.session.scalars(table.limit(1)).first() def roundHalfUpToInt(number): return int(Decimal(number).quantize(Decimal(0), rounding=ROUND_HALF_UP)) class Student(db.Model): id = db.Column(db.Integer, primary_key=True) student_number = db.Column( db.Integer, db.CheckConstraint(f"student_number >= {MIN_STUDENT_NUMBER}"), nullable=False, unique=True ) first_name = db.Column(db.String(100), nullable=False) last_name = db.Column(db.String(100), nullable=False) uni_email = db.Column(db.String(200), nullable=False, unique=True) contact_email = db.Column(db.String(200), nullable=True, unique=True) bachelor_thesis = db.Column(db.String, nullable=True) bachelor_thesis_work_group = db.Column(db.String, nullable=True) note = db.Column(db.Text, nullable=True) part_students = db.relationship("PartStudent", back_populates="student", lazy=True) def __init__(self, uni_email, contact_email=None, **kwargs): # Lower and strip uni email uni_email = uni_email.strip().lower() if contact_email is not None: # Lower and strip contact email contact_email = contact_email.strip().lower() # Don't save contact_email if it is similar to uni_email if contact_email == uni_email: contact_email = None super().__init__(uni_email=uni_email, contact_email=contact_email, **kwargs) def str(self): return f"{self.first_name} {self.last_name}" def __str__(self): return self.str() class PartStudent(db.Model): # A student doing a specific part id = db.Column(db.Integer, primary_key=True) final_part_mark = db.Column( db.Integer, db.CheckConstraint(f"final_part_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"), nullable=True, ) student_id = db.Column(db.Integer, db.ForeignKey("student.id"), nullable=False) student = db.relationship("Student", back_populates="part_students") part_id = db.Column(db.Integer, db.ForeignKey("part.id"), nullable=False) part = db.relationship("Part", back_populates="part_students") group_id = db.Column(db.Integer, db.ForeignKey("group.id"), nullable=True) group = db.relationship("Group", back_populates="part_students") experiment_marks = db.relationship("ExperimentMark", back_populates="part_student", lazy=True) __table_args__ = ( db.UniqueConstraint(student_id, part_id), db.UniqueConstraint(student_id, group_id), ) def str(self): return f"{self.student.str()} {self.part}" def __str__(self): return f"<{self.str()}>" @staticmethod def check(part, group): if group is not None and group.program != part.program: raise DatabaseException( f"Group's program {group.program} and student part's program {part.program} do not match!" ) def __init__(self, part, group=None, **kwargs): PartStudent.check(part, group) super().__init__(part=part, group=group, **kwargs) def checkThenSetFinalPartMark(self): """ Return True if final_part_mark changed, False otherwise. """ finalWeightingSum = 0 finalMarkSum = 0 groupExperiments = [] for experimentMark in self.experiment_marks: if None in (experimentMark.oral_mark, experimentMark.protocol_mark): # Not all marks are set! try: self.final_part_mark = None db.session.commit() return True except Exception as ex: flash(str(ex), "error") db.session.rollback() return False groupExperiment = experimentMark.group_experiment groupExperiments.append(groupExperiment) semesterExperiment = groupExperiment.semester_experiment finalWeighting = semesterExperiment.final_weighting finalWeightingSum += finalWeighting # Not using final_experiment_mark to avoid rounding two times finalMarkSum += finalWeighting * ( semesterExperiment.protocol_weighting * experimentMark.protocol_mark + semesterExperiment.oral_weighting * experimentMark.oral_mark ) if set(groupExperiments) != set(self.group.group_experiments): flash(f"{self} does not have an experiment mark for every group experiment in his group!", "warning") return False oldFinalPartMark = self.final_part_mark try: self.final_part_mark = roundHalfUpToInt(finalMarkSum / finalWeightingSum) db.session.commit() except Exception as ex: flash(str(ex), "error") db.session.rollback() return False # Inform admin about what changed if current_user.has_role("admin"): if oldFinalPartMark != self.final_part_mark: category = "danger" if oldFinalPartMark and oldFinalPartMark > self.final_part_mark else "info" flash( f"Final part mark changed for {self} from {oldFinalPartMark} to {self.final_part_mark}.", category, ) else: flash(f"Final part mark did not change for {self} from {oldFinalPartMark}.", "warning") return True class Group(db.Model): id = db.Column(db.Integer, primary_key=True) number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_GROUP_NUMBER}"), nullable=False) semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False) semester = db.relationship("Semester", back_populates="groups") program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False) program = db.relationship("Program", back_populates="groups") part_students = db.relationship("PartStudent", back_populates="group", lazy=True) group_experiments = db.relationship("GroupExperiment", back_populates="group", lazy=True) __table_args__ = (db.UniqueConstraint(number, semester_id, program_id),) def str_without_semester(self): return f"{self.number} {self.program.str()}" def str(self): return f"{self.str_without_semester()} {self.semester.str()}" def __str__(self): return f"<{self.str()}>" @staticmethod def check(part_students, program=None): commonProgram = part_students[0].part.program if program and program != commonProgram: raise DatabaseException("Group's program and students' program mismatch!") for partStudent in part_students[1:]: if partStudent.part.program != commonProgram: raise DatabaseException(f"Part Students {part_students} are not in the same program!") @staticmethod def customInit(part_students): Group.check(part_students) semester = part_students[0].part.semester program = part_students[0].part.program highestGroupNumber = get_first( select(Group.number) .where(Group.semester == semester, Group.program == program) .order_by(Group.number.desc()) ) number = highestGroupNumber + 1 if highestGroupNumber is not None else 1 return Group( program=program, number=number, part_students=part_students, semester=semester, ) class GroupExperiment(db.Model): # An experiment specified to a group id = db.Column(db.Integer, primary_key=True) note = db.Column(db.Text, nullable=True) experiment_marks_missing = db.Column(db.Boolean, default=True, nullable=False) semester_experiment_id = db.Column(db.Integer, db.ForeignKey("semester_experiment.id"), nullable=False) semester_experiment = db.relationship("SemesterExperiment", back_populates="group_experiments") group_id = db.Column(db.Integer, db.ForeignKey("group.id"), nullable=False) group = db.relationship("Group", back_populates="group_experiments") appointments = db.relationship("Appointment", back_populates="group_experiment", lazy=True) experiment_marks = db.relationship("ExperimentMark", back_populates="group_experiment", lazy=True) __table_args__ = (db.UniqueConstraint(semester_experiment_id, group_id),) def str(self): return f"" def __str__(self): return self.str() @staticmethod def check(semester_experiment, group): for partStudent in group.part_students: student = partStudent.student for partStudent in student.part_students: for experimentMark in partStudent.experiment_marks: if experimentMark.group_experiment.semester_experiment.experiment == semester_experiment.experiment: raise DatabaseException( f"{student} has already done {semester_experiment.experiment} in {partStudent.part} and had {experimentMark}!" ) def __init__(self, semester_experiment, group, **kwargs): GroupExperiment.check(semester_experiment, group) super().__init__(semester_experiment=semester_experiment, group=group, **kwargs) for partStudent in group.part_students: db.session.add(ExperimentMark(part_student=partStudent, group_experiment=self)) def update_experiment_marks_missing(self): for experiment_mark in self.experiment_marks: if experiment_mark.final_experiment_mark is None: self.experiment_marks_missing = True return self.experiment_marks_missing = False class Experiment(db.Model): id = db.Column(db.Integer, primary_key=True) number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_EXPERIMENT_NUMBER}"), nullable=False) title = db.Column(db.String(200), nullable=False) description = db.Column(db.Text, nullable=True) wiki_link = db.Column(db.String(300), nullable=True) building = db.Column(db.String(100), nullable=False) room = db.Column(db.String(100), nullable=False) responsibility = db.Column(db.String(200), nullable=True) duration_in_days = db.Column( db.Integer, db.CheckConstraint(f"duration_in_days >= {MIN_DURATION_IN_DAYS}"), nullable=False ) active = db.Column(db.Boolean, default=True, nullable=False) program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False) program = db.relationship("Program", back_populates="experiments") semester_experiments = db.relationship("SemesterExperiment", back_populates="experiment", lazy=True) __table_args__ = (db.UniqueConstraint(number, program_id),) def str(self): return f"{self.number} {self.program.str()}" def __str__(self): return f"<{self.str()}>" # Helper table for the many to many relationship between Assistant and SemesterExperiment experiment_assistant = db.Table( "experiment_assistant", db.Column("semester_experiment_id", db.Integer, db.ForeignKey("semester_experiment.id"), primary_key=True), db.Column("assistant_id", db.Integer, db.ForeignKey("assistant.id"), primary_key=True), ) class SemesterExperiment(db.Model): # An experiment in a specific semester id = db.Column(db.Integer, primary_key=True) oral_weighting = db.Column( db.Float, db.CheckConstraint("oral_weighting BETWEEN 0 AND 1"), default=0.5, nullable=False, ) protocol_weighting = db.Column( db.Float, db.CheckConstraint("protocol_weighting BETWEEN 0 AND 1"), default=0.5, nullable=False, ) final_weighting = db.Column( db.Float, db.CheckConstraint("final_weighting BETWEEN 0 AND 1"), default=1.0, nullable=False, ) experiment_id = db.Column(db.Integer, db.ForeignKey("experiment.id"), nullable=False) experiment = db.relationship("Experiment", back_populates="semester_experiments") semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False) semester = db.relationship("Semester", back_populates="semester_experiments") assistants = db.relationship( "Assistant", secondary=experiment_assistant, back_populates="semester_experiments", lazy=True ) group_experiments = db.relationship("GroupExperiment", back_populates="semester_experiment", lazy=True) __table_args__ = (db.UniqueConstraint(experiment_id, semester_id),) def str_without_semester(self): return f"{self.experiment.str()}" def str(self): return f"{self.str_without_semester()} {self.semester.str()}" def __str__(self): return f"<{self.str()}>" def checkAndRoundWeightings(self): roundedOralWeighting = round(self.oral_weighting, 2) roundedProtocolWeighting = round(self.protocol_weighting, 2) weightingSum = round(roundedOralWeighting + roundedProtocolWeighting, 2) if weightingSum != 1: raise DatabaseException( f"Oral and protocol weightings (rounded to 2 decimal digits) sum to {weightingSum} and not 1.00!" ) self.oral_weighting = roundedOralWeighting self.protocol_weighting = roundedProtocolWeighting def updateFinalExperimentAndPartMarks(self): for groupExperiment in self.group_experiments: for experimentMark in groupExperiment.experiment_marks: experimentMark.update_final_experiment_mark() for groupExperiment in self.group_experiments: for partStudent in groupExperiment.group.part_students: partStudent.checkThenSetFinalPartMark() class Assistant(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False, unique=True) user = db.relationship("User", back_populates="assistant") semester_experiments = db.relationship( "SemesterExperiment", secondary=experiment_assistant, back_populates="assistants", lazy=True ) appointments = db.relationship("Appointment", back_populates="assistant", lazy=True) experiment_marks = db.relationship("ExperimentMark", back_populates="assistant", lazy=True) def str(self): return f"{self.user}" def __str__(self): return self.str() class Admin(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False, unique=True) user = db.relationship("User", back_populates="admin") experiment_marks = db.relationship("ExperimentMark", back_populates="admin", lazy=True) def str(self): return f"{self.user}" def __str__(self): return self.str() class Appointment(db.Model): id = db.Column(db.Integer, primary_key=True) date = db.Column( db.Date, db.CheckConstraint(f"date BETWEEN '20{MIN_YEAR}-01-01' AND '20{MAX_YEAR}-01-01'"), nullable=False ) # To be specified with the python package "datetime" special = db.Column(db.Boolean, default=False, nullable=False) # In the break or not group_experiment_id = db.Column(db.Integer, db.ForeignKey("group_experiment.id"), nullable=False) group_experiment = db.relationship("GroupExperiment", back_populates="appointments") assistant_id = db.Column(db.Integer, db.ForeignKey("assistant.id"), nullable=False) assistant = db.relationship("Assistant", back_populates="appointments") def str(self): return f"{self.date} {self.group_experiment.str()}" def __str__(self): return f"<{self.str()}>" @staticmethod def checkAndGetAssistant(groupExperiment, assistant=None): semesterExperiment = groupExperiment.semester_experiment semesterExperimentAssistants = semesterExperiment.assistants if semesterExperimentAssistants is None: raise DatabaseException(f"{semesterExperiment} does not have assistants yet!") if assistant is not None: if assistant not in semesterExperimentAssistants: raise DatabaseException(f"{assistant} is not responsible for {semesterExperiment}!") elif len(semesterExperimentAssistants) == 1: assistant = semesterExperimentAssistants[0] else: raise DatabaseException( f"Experiment {semesterExperiment} has more than one assistant. You have to assign one of these assistants: {semesterExperimentAssistants}" ) return assistant def __init__(self, group_experiment, assistant=None, **kwargs): assistant = Appointment.checkAndGetAssistant(group_experiment, assistant) super().__init__(group_experiment=group_experiment, assistant=assistant, **kwargs) def custom_update(self, date, special, group_experiment, assistant=None): self.assistant = Appointment.checkAndGetAssistant(group_experiment, assistant) self.date = date self.special = special self.group_experiment = group_experiment class Part(db.Model): id = db.Column(db.Integer, primary_key=True) number = db.Column(db.Integer, db.CheckConstraint(f"number >= {MIN_PART_NUMBER}"), nullable=False) program_id = db.Column(db.Integer, db.ForeignKey("program.id"), nullable=False) program = db.relationship("Program", back_populates="parts") semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False) semester = db.relationship("Semester", back_populates="parts") part_students = db.relationship("PartStudent", back_populates="part", lazy=True) __table_args__ = (db.UniqueConstraint(program_id, number, semester_id),) def str_without_semester(self): return f"{self.program.str()}{self.number}" def str(self): return f"{self.str_without_semester()} {self.semester.str()}" def __str__(self): return f"<{self.str()}>" class Semester(db.Model): id = db.Column(db.Integer, primary_key=True) label = db.Column(db.String(10), db.CheckConstraint("label IN ('WS', 'SS')"), nullable=False) year = db.Column(db.Integer, db.CheckConstraint(f"year BETWEEN {MIN_YEAR} AND {MAX_YEAR}"), nullable=False) # Assistants can not work in semesters that are done done = db.Column(db.Boolean, default=False, nullable=False) parts = db.relationship("Part", back_populates="semester", lazy=True) semester_experiments = db.relationship("SemesterExperiment", back_populates="semester", lazy=True) active_users = db.relationship("User", back_populates="active_semester", lazy=True) groups = db.relationship("Group", back_populates="semester", lazy=True) __table_args__ = (db.UniqueConstraint(label, year),) def str(self): return f"{self.label}{self.year}" def __str__(self): return self.str() def __init__(self, label, year, **kwargs): last_semester = Semester.lastSemester() if last_semester is not None and (year < last_semester.year or (year == last_semester.year and label == "SS")): raise DatabaseException(f"You can only create semesters later than the last semester {last_semester}!") super().__init__(label=label, year=year, **kwargs) @staticmethod def initFromOldSemester(label, year, oldSemester, transferParts, transferAssistants): semester = Semester(label=label, year=year) if transferParts: semester.transferPartsFrom(oldSemester) for experiment in db.session.scalars(select(Experiment).where(Experiment.active == True)): newSemesterExperiment = SemesterExperiment(experiment=experiment, semester=semester) if transferAssistants: for oldSemesterExperiment in oldSemester.semester_experiments: if oldSemesterExperiment.experiment == experiment: newSemesterExperiment.assistants = oldSemesterExperiment.assistants db.session.add(newSemesterExperiment) return semester def transferPartsFrom(self, oldSemester): for part in oldSemester.parts: db.session.add(Part(program=part.program, number=part.number, semester=self)) @staticmethod def sortedSemestersStartingWithNewest(limit=0): # Inserting an older semester is not allowed! # Therefore, the id is enough. stmt = select(Semester).order_by(Semester.id.desc()) if limit > 0: stmt = stmt.limit(limit) return db.session.scalars(stmt) @staticmethod def lastSemester(): return Semester.sortedSemestersStartingWithNewest(limit=1).first() def num_missing_experiment_marks(self): return db.session.scalar( select(func.count()) .select_from(ExperimentMark) .join(GroupExperiment) .join(SemesterExperiment) .where(SemesterExperiment.semester == self) .where(ExperimentMark.final_experiment_mark == None) ) def set_done(self, next_semester=None): set_next_semester = next_semester is not None # Set also all previous semesters as done for id in range(1, self.id + 1): semester = db.session.get(Semester, id) if semester == self or not semester.done: num_missing_experiment_marks = self.num_missing_experiment_marks() if num_missing_experiment_marks > 0: flash( f"Semester {semester} was set as done, but it has {num_missing_experiment_marks} missing experiment marks!", "danger", ) semester.done = True if set_next_semester: # Set active_semester to next_semester users_in_semester_done = db.session.scalars(select(User).where(User.active_semester == semester)) for user in users_in_semester_done: user.active_semester = next_semester if user == current_user: flash(f"Active semester changed to the next semester {next_semester}!", "warning") class ExperimentMark(db.Model): # A mark for a student after a specific experiment id = db.Column(db.Integer, primary_key=True) oral_mark = db.Column( db.Integer, db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"), nullable=True, ) protocol_mark = db.Column( db.Integer, db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"), nullable=True, ) final_experiment_mark = db.Column( db.Integer, db.CheckConstraint(f"oral_mark BETWEEN {MIN_MARK} AND {MAX_MARK}"), nullable=True, ) part_student_id = db.Column(db.Integer, db.ForeignKey("part_student.id"), nullable=False) part_student = db.relationship("PartStudent", back_populates="experiment_marks") group_experiment_id = db.Column(db.Integer, db.ForeignKey("group_experiment.id"), nullable=False) group_experiment = db.relationship("GroupExperiment", back_populates="experiment_marks") assistant_id = db.Column( db.Integer, db.ForeignKey("assistant.id"), nullable=True ) # The last assistant who edited the mark assistant = db.relationship("Assistant", back_populates="experiment_marks") admin_id = db.Column(db.Integer, db.ForeignKey("admin.id"), nullable=True) # The last admin who edited the mark admin = db.relationship("Admin", back_populates="experiment_marks") __table_args__ = (db.UniqueConstraint(part_student_id, group_experiment_id),) def str(self): return ( f"" ) def __str__(self): return self.str() @staticmethod def check(part_student, group_experiment): if not part_student.group: raise DatabaseException("The part student does not have a group yet!") if group_experiment not in part_student.group.group_experiments: raise DatabaseException("The group of the part student does not have the given group experiment!") def __init__(self, part_student, group_experiment, **kwargs): ExperimentMark.check(part_student, group_experiment) super().__init__(part_student=part_student, group_experiment=group_experiment, **kwargs) def update_final_experiment_mark(self): """ Return True if final_experiment_mark changed, False otherwise. """ old_final_experiment_mark = self.final_experiment_mark oral_mark = self.oral_mark protocol_mark = self.protocol_mark if None in (oral_mark, protocol_mark): self.final_experiment_mark = None else: semesterExperiment = self.group_experiment.semester_experiment self.final_experiment_mark = roundHalfUpToInt( semesterExperiment.oral_weighting * oral_mark + semesterExperiment.protocol_weighting * protocol_mark ) final_experiment_mark_changed = self.final_experiment_mark != old_final_experiment_mark return final_experiment_mark_changed def set_oral_protocol_mark(self, oral_mark: int, protocol_mark: int, call_update_experiment_marks_missing=True): """ Return True if final_experiment_mark changed, False otherwise. """ self.oral_mark = oral_mark self.protocol_mark = protocol_mark final_experiment_mark_changed = self.update_final_experiment_mark() if final_experiment_mark_changed: self.part_student.checkThenSetFinalPartMark() if call_update_experiment_marks_missing: self.group_experiment.update_experiment_marks_missing() return final_experiment_mark_changed class User(db.Model, FsUserMixin): first_name = db.Column(db.String(100), nullable=False) last_name = db.Column(db.String(100), nullable=False) phone_number = db.Column(db.String(50), nullable=True) mobile_phone_number = db.Column(db.String(50), nullable=True) building = db.Column(db.String(100), nullable=True) room = db.Column(db.String(100), nullable=True) active_semester_id = db.Column(db.Integer, db.ForeignKey("semester.id"), nullable=False) active_semester = db.relationship("Semester", back_populates="active_users") admin = db.relationship("Admin", back_populates="user", lazy=False, uselist=False) assistant = db.relationship("Assistant", back_populates="user", lazy=True, uselist=False) def set_last_semester_as_active(self): """ Return True if changed, False otherwise. """ last_semester = Semester.lastSemester() if last_semester.done: return False try: self.active_semester = last_semester db.session.commit() except Exception as ex: flash(str(ex), "error") db.session.rollback() return False else: flash( f"Active semester changed to {last_semester} because your last active semester was set as done!", "warning", ) return True def str(self): return f"{self.first_name} {self.last_name}" def __str__(self): return self.str() class Role(db.Model, FsRoleMixin): def str(self): return f"{self.name}" def __str__(self): return self.str() class Program(db.Model): id = db.Column(db.Integer, primary_key=True) label = db.Column(db.String(25), nullable=False, unique=True) parts = db.relationship("Part", back_populates="program", lazy=True) experiments = db.relationship("Experiment", back_populates="program", lazy=True) groups = db.relationship("Group", back_populates="program", lazy=True) def str(self): return f"{self.label}" def __str__(self): return self.str()