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 Form, BooleanField, SelectField, TextField, RadioField, FloatField
from wtforms.validators import DataRequired, Email, Optional
from flask_admin.contrib.sqla.fields import QuerySelectMultipleField, QuerySelectField
from flask_admin.helpers import get_form_data

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 UserView(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.on_model_change(form, model, True)
            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 RoleView(SecureModelView):
    can_create = False
    can_edit = False
    can_delete = False
    column_display_actions = False

    column_list = ["name", "description"]


class SemesterView(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.on_model_change(form, model, True)
            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 PartView(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 StudentView(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_sortable_list = ["student_number", "first_name", "last_name"]
    column_searchable_list = column_sortable_list + ["uni_email", "contact_email"]

    form_excluded_columns = ["part_students"]

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


def partQueryFactory():
    return Part.query.filter(Part.id.in_([part.id for part in userActiveSemester().parts]))


class PartStudentView(SecureModelView):
    class CreateForm(Form):
        def studentQueryFactory():
            return Student.query

        def groupQueryFactory():
            return Group.query.filter(Group.part_id.in_([part.id for part in userActiveSemester().parts]))

        student = QuerySelectField(
            "Student", query_factory=studentQueryFactory, validators=[DataRequired()], allow_blank=True, blank_text="-"
        )
        part = QuerySelectField(
            "Part", query_factory=partQueryFactory, validators=[DataRequired()], allow_blank=True, blank_text="-"
        )
        group = QuerySelectField("Group", query_factory=groupQueryFactory, allow_blank=True, blank_text="-")

    class EditForm(CreateForm):
        student = None
        part = None
        final_part_mark = FloatField("Final Part Mark", validators=[Optional()])

    form = EditForm

    column_filters = ["part", "student", "group"]

    partGroupPartMismatchException = "Part and groups part don't match!"

    def create_form(self, obj=None):
        form = self.CreateForm
        return form(get_form_data(), obj=obj)

    def on_model_change(self, form, model, is_created):
        if model.group and model.part != model.group.part:
            raise Exception(self.partGroupPartMismatchException)

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

    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 GroupView(SecureModelView):
    class CreateForm(Form):
        def partStudentsQueryFactory():
            return PartStudent.query.filter(PartStudent.part_id.in_([part.id for part in userActiveSemester().parts]))

        part = QuerySelectField(
            label="Part", query_factory=partQueryFactory, validators=[DataRequired()], allow_blank=True, blank_text="-"
        )
        part_students = QuerySelectMultipleField(label="Part Students", query_factory=partStudentsQueryFactory)

    class EditForm(CreateForm):
        part = None

    form = EditForm

    column_list = ["number", "part", "part_students"]
    column_filters = ["part"]
    column_searchable_list = ["number"]

    partStudentPartPartMismatchException = "Part and StudentParts part don't match!"

    def create_model(self, form):
        try:
            orderedPartGroups = Group.query.filter(Group.part == form.part.data).order_by(Group.number)
            lastTakenGroupNumber = orderedPartGroups[-1].number if orderedPartGroups.count() > 0 else 0

            model = Group(
                number=lastTakenGroupNumber + 1,
                part_students=form.part_students.data,
                part=form.part.data,
            )

            self.session.add(model)
            self.on_model_change(form, model, True)
            self.session.commit()
        except Exception as ex:
            flash(ex, "error")

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

    def on_model_change(self, form, model, is_created):
        for partStudent in model.part_students:
            if model.part != partStudent.part:
                raise Exception(self.partStudentPartPartMismatchException)

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

    def create_form(self, obj=None):
        form = self.CreateForm
        return form(get_form_data(), obj=obj)

    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 ExperimentView(SecureModelView):
    can_view_details = True
    column_filters = ["deprecated"]
    column_list = ["number", "name", "deprecated"]


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

    partLabels = getConfig("partLabels")
    column_filters = ["part"]

    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 AssistantView(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 GroupExperimentView(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(StudentView(Student, db.session))
admin.add_view(PartStudentView(PartStudent, db.session))
admin.add_view(GroupView(Group, db.session))
admin.add_view(GroupExperimentView(GroupExperiment, db.session))
admin.add_view(ExperimentView(Experiment, db.session))
admin.add_view(PartExperimentView(PartExperiment, db.session))
admin.add_view(AssistantView(Assistant, db.session))
admin.add_view(SecureModelView(Appointment, db.session))
admin.add_view(PartView(Part, db.session))
admin.add_view(SemesterView(Semester, db.session))
admin.add_view(SecureModelView(ExperimentMark, db.session))
admin.add_view(UserView(User, db.session))
admin.add_view(RoleView(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",
            )
        )

    admin.add_link(MenuLink(name="Logout", url=url_for("security.logout")))