AdvLabDB/advlabdb/modelViews.py

from flask import flash, request, url_for
from flask_admin.contrib.sqla.filters import BaseSQLAFilter
from flask_admin.menu import MenuLink
from flask_security import current_user, hash_password
from sqlalchemy import func
from wtforms import BooleanField, SelectField, TextField, RadioField
from wtforms.validators import DataRequired, Email

from advlabdb import admin, app, db, user_datastore
from advlabdb.configUtils import getConfig
from advlabdb.customClasses import SecureModelView
from advlabdb.models import (
    Appointment,
    Assistant,
    Experiment,
    ExperimentMark,
    Group,
    GroupExperiment,
    Part,
    PartExperiment,
    PartStudent,
    Role,
    Semester,
    Student,
    User,
)
from advlabdb.utils import (
    partFromLabelInUserActiveSemester,
    randomPassword,
    setUserActiveSemester,
    userActiveSemester,
)


class UserModelView(SecureModelView):
    column_list = ["email", "active", "roles", "assistant"]
    column_searchable_list = ["email"]
    column_filters = ["active"]
    form_columns = ["email", "active", "roles"]

    form_args = {
        "email": {"validators": [Email()]},
        "active": {"default": True},
        "roles": {"validators": [DataRequired(message="A role is required!")]},
    }

    deleteSelfException = "Tried to delete yourself as user!"
    deactivateSelfException = "Tried to deactiavte yourself as user!"

    def create_model(self, form):
        password = randomPassword()
        passwordHash = hash_password(password)

        email = form.email.data.lower()

        roles = [role.name for role in form.roles.data]
        if "admin" in roles:
            flash("You have registered a new admin!", "danger")

        try:
            model = user_datastore.create_user(email=email, password=passwordHash, roles=roles)
            self.session.commit()
        except Exception as ex:
            flash(ex, "error")

            self.session.rollback()
        else:
            flash(
                f"{email} registered with roles: {', '.join([role.name for role in form.roles.data])}.",
                category="success",
            )
            flash(f"Random password: {password}", category="warning")
            return model

    def on_model_delete(self, model):
        if model == current_user:
            raise Exception(self.deleteSelfException)

    def on_model_change(self, form, model, is_created):
        if model == current_user and not form.active.data:
            raise Exception(self.deactivateSelfException)

    def handle_view_exception(self, exc):
        if exc.args[0] in (self.deleteSelfException, self.deactivateSelfException):
            pass
        else:
            return super().handle_view_exception(exc)


class RoleModelView(SecureModelView):
    can_create = False
    can_edit = False
    can_delete = False
    column_display_actions = False

    column_list = ["name", "description"]


class SemesterModelView(SecureModelView):
    can_edit = False

    column_list = ["label", "parts"]
    form_columns = ["semester_label", "year", "create_parts", "transfer_assistants"]

    semesterLabels = ["WS", "SS"]
    form_extra_fields = {
        "semester_label": RadioField(
            "Semester", choices=list(zip(semesterLabels, semesterLabels)), validators=[DataRequired()]
        ),
        "year": TextField("Year", validators=[DataRequired()]),
        "create_parts": BooleanField(
            "Create parts:"
            + ", ".join(getConfig("partLabels"))
            + " and transfer part experiments from your current active semester:",
            default=True,
        ),
        "transfer_assistants": BooleanField("Transfer Assistants from your current active semester:", default=False),
    }

    def create_model(self, form):
        try:
            model = Semester(label=form.semester_label.data + form.year.data)
            self.session.add(model)
            self.session.commit()
        except Exception as ex:
            flash(ex, "error")

            self.session.rollback()
        else:
            self.after_model_change(form, model, True)
            return model

    def after_model_change(self, form, model, is_created):
        admin.add_link(
            MenuLink(
                name=model.label,
                url=url_for("set_semester") + "?semester_id=" + str(model.id),
                category="Active semester",
            )
        )

        oldSemesterParts = userActiveSemester().parts

        setUserActiveSemester(model.id)

        if form.create_parts.data:
            model.createParts()

            try:
                for part in oldSemesterParts:
                    for partExperiment in part.part_experiments:
                        newPartExperiment = PartExperiment(
                            experiment=partExperiment.experiment, part=partFromLabelInUserActiveSemester(part.label)
                        )
                        if form.transfer_assistants.data:
                            newPartExperiment.assistants = partExperiment.assistants
                        self.session.add(newPartExperiment)
                self.session.commit()
            except Exception as ex:
                flash(ex, "error")

                self.session.rollback()


class PartModelView(SecureModelView):
    can_view_details = True
    column_details_list = ["label", "semester", "part_experiments", "part_students", "groups"]
    form_columns = ["label", "semester"]

    partLabels = getConfig("partLabels")
    form_choices = {"label": list(zip(partLabels, partLabels))}

    def get_query(self):
        return super().get_query().filter(Part.id.in_([part.id for part in userActiveSemester().parts]))

    def get_count_query(self):
        return (
            self.session.query(func.count("*"))
            .select_from(self.model)
            .filter(Part.id.in_([part.id for part in userActiveSemester().parts]))
        )


class StudentModelView(SecureModelView):
    can_view_details = True
    column_list = ["student_number", "first_name", "last_name", "uni_email", "contact_email", "part_students"]
    column_details_list = column_list + ["bachelor_thesis", "bachelor_thesis_work_group", "note"]
    column_searchable_list = ["student_number", "uni_email", "contact_email", "first_name", "last_name"]
    form_columns = column_details_list + ["new_part_student_part", "new_part_student_group_number"]

    form_args = {
        "uni_email": {"validators": [Email()]},
        "contact_email": {"validators": [Email()]},
    }

    partChoices = ["-"] + getConfig("partLabels")
    form_extra_fields = {
        "new_part_student_part": SelectField(
            "Part", choices=list(zip(partChoices, partChoices)), default=partChoices[0]
        ),
        "new_part_student_group_number": TextField("Group number"),
    }

    def validate_form(self, form):
        if request.method == "POST":
            partLabel = form.new_part_student_part.data
            groupNumber = form.new_part_student_group_number.data

            if (partLabel != self.partChoices[0] and groupNumber == "") or (
                partLabel == self.partChoices[0] and groupNumber != ""
            ):
                flash("You have to assign both part and group if you want to add a part student!", "danger")
                return False

            if partLabel != self.partChoices[0] and not partFromLabelInUserActiveSemester(partLabel):
                flash(f"Part {partLabel} is not created in {str(userActiveSemester())} yet!", "danger")
                return False

            if groupNumber != "":
                message = "The group number has to be an integer > 0 !"

                try:
                    groupNumber = int(groupNumber)
                except Exception:
                    flash(message, "danger")
                    return False

                if groupNumber < 1:
                    flash(message, "danger")
                    return False

        return super().validate_form(form)

    def after_model_change(self, form, model, is_created):
        partLabel = form.new_part_student_part.data

        if partLabel != self.partChoices[0]:
            groupNumber = int(form.new_part_student_group_number.data)

            part = partFromLabelInUserActiveSemester(partLabel)
            group = Group.query.filter(Group.number == groupNumber, Group.part == part).first()

            if group is None:
                try:
                    group = Group(number=groupNumber, part=part)
                    self.session.add(group)
                    self.session.commit()
                except Exception as ex:
                    flash(ex, "error")

                    self.session.rollback()
                    return
                else:
                    flash(f"Added the new group with number {str(groupNumber)} in part {str(part)}.", "success")

            try:
                partStudent = PartStudent(student=model, part=part, group=group)
                self.session.add(partStudent)
                self.session.commit()
            except Exception as ex:
                flash(ex, "error")

                self.session.rollback()
            else:
                flash("Added part student.", "success")


class PartFilter(BaseSQLAFilter):
    def apply(self, query, value, alias=None):
        return query.filter(self.column == partFromLabelInUserActiveSemester(value).id)

    def operation(self):
        return "equals"

    def validate(self, value):
        if partFromLabelInUserActiveSemester(value):
            return True
        else:
            flash(f"Part {value} not found in your active semester {userActiveSemester()}!", "danger")
            return False


class PartStudentModelView(SecureModelView):
    partLabels = getConfig("partLabels")
    column_filters = [PartFilter(PartStudent.part_id, "Part", options=list(zip(partLabels, partLabels)))]

    form_excluded_columns = ["experiment_marks"]

    def get_query(self):
        return super().get_query().filter(PartStudent.part_id.in_([part.id for part in userActiveSemester().parts]))

    def get_count_query(self):
        return (
            self.session.query(func.count("*"))
            .select_from(self.model)
            .filter(PartStudent.part_id.in_([part.id for part in userActiveSemester().parts]))
        )


class GroupModelView(SecureModelView):
    partLabels = getConfig("partLabels")
    column_filters = [PartFilter(Group.part_id, "Part", options=list(zip(partLabels, partLabels)))]

    def validate_form(self, form):
        if request.method == "POST":
            if Group.query.filter(Group.number == form.number.data, Group.part == form.part.data).first():
                lastTakenGroupNumber = (
                    Group.query.filter(Group.part == form.part.data).order_by(Group.number)[-1].number
                )
                flash(
                    f"Group number taken in this part! Last group number taken in this part is {lastTakenGroupNumber}.",
                    "warning",
                )
                return False
        return super().validate_form(form)

    def get_query(self):
        return super().get_query().filter(Group.part_id.in_([part.id for part in userActiveSemester().parts]))

    def get_count_query(self):
        return (
            self.session.query(func.count("*"))
            .select_from(self.model)
            .filter(Group.part_id.in_([part.id for part in userActiveSemester().parts]))
        )


class ExperimentModelView(SecureModelView):
    can_view_details = True
    column_filters = ["deprecated"]
    column_list = ["number", "name", "deprecated"]


class PartExperimentModelView(SecureModelView):
    column_list = ["experiment", "part", "assistants"]

    partLabels = getConfig("partLabels")
    column_filters = [PartFilter(PartExperiment.part_id, "Part", options=list(zip(partLabels, partLabels)))]

    def get_query(self):
        return super().get_query().filter(PartExperiment.part_id.in_([part.id for part in userActiveSemester().parts]))

    def get_count_query(self):
        return (
            self.session.query(func.count("*"))
            .select_from(self.model)
            .filter(PartExperiment.part_id.in_([part.id for part in userActiveSemester().parts]))
        )


class AssistantModelView(SecureModelView):
    can_view_details = True
    column_list = ["first_name", "last_name", "email", "user", "part_experiments"]
    column_details_list = column_list + ["phone_number", "mobile_phone_number", "room", "building", "experiment_marks"]
    column_filters = ["user.active"]
    form_excluded_columns = ["experiment_marks"]


class GroupExperimentModelView(SecureModelView):
    column_list = ["group", "part_experiment", "appointments", "experiment_marks"]

    def get_query(self):
        return (
            super()
            .get_query()
            .filter(
                GroupExperiment.group_id.in_(
                    [
                        g.id
                        for g in Group.query.filter(Group.part_id.in_([part.id for part in userActiveSemester().parts]))
                    ]
                )
            )
        )

    def get_count_query(self):
        return (
            self.session.query(func.count("*"))
            .select_from(self.model)
            .filter(
                GroupExperiment.group_id.in_(
                    [
                        g.id
                        for g in Group.query.filter(Group.part_id.in_([part.id for part in userActiveSemester().parts]))
                    ]
                )
            )
        )


admin.add_view(StudentModelView(Student, db.session))
admin.add_view(PartStudentModelView(PartStudent, db.session))
admin.add_view(GroupModelView(Group, db.session))
admin.add_view(GroupExperimentModelView(GroupExperiment, db.session))
admin.add_view(ExperimentModelView(Experiment, db.session))
admin.add_view(PartExperimentModelView(PartExperiment, db.session))
admin.add_view(AssistantModelView(Assistant, db.session))
admin.add_view(SecureModelView(Appointment, db.session))
admin.add_view(PartModelView(Part, db.session))
admin.add_view(SemesterModelView(Semester, db.session))
admin.add_view(SecureModelView(ExperimentMark, db.session))
admin.add_view(UserModelView(User, db.session))
admin.add_view(RoleModelView(Role, db.session))

with app.app_context():
    semesters = Semester.query.all()[::-1]
    for semester in semesters:
        admin.add_link(
            MenuLink(
                name=semester.label,
                url=url_for("set_semester") + "?semester_id=" + str(semester.id),
                category="Active semester",
            )
        )