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_admin.model.template import EndpointLinkRowAction
from flask_security import current_user, hash_password
from sqlalchemy import func
from wtforms import Form, BooleanField, SelectField, TextField, RadioField
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 wtforms.fields.html5 import DateField

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,
    SemesterExperiment,
    PartStudent,
    Role,
    Semester,
    Student,
    User,
)
from advlabdb.utils import (
    partFromLabelInUserActiveSemester,
    randomPassword,
    setUserActiveSemester,
    userActiveSemester,
)


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

    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(str(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
    can_delete = False
    can_view_details = True

    column_list = ["label", "parts"]
    column_details_list = column_list + ["active_users"]
    form_columns = ["semester_label", "year", "transfer_parts", "transfer_assistants"]

    form_extra_fields = {
        "semester_label": RadioField("Semester", choices=["WS", "SS"], validators=[DataRequired()]),
        "year": TextField("Year", validators=[DataRequired()]),
        "transfer_parts": BooleanField(
            "Transfer parts",
            description="This option transfers the parts you have in your current active semester. Make sure that your semester is the last semester before creating a new one (recommended)!",
            default=True,
        ),
        "transfer_assistants": BooleanField(
            "Transfer Assistants",
            description="This option transfers assistants of your active semester to active experiments in the new semester. Make sure that your semester is the last semester before creating a new one (recommended)! Active experiments are transfered anyway. If you do not want an experiment to be transfered, set it to inactive before creating the new semester. Experiments which are switched to active before creating the new semester will be created in the new semester without assistants. It is important to check the assistants of all experiments after creating a new semester.",
            default=True,
        ),
    }

    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(str(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",
            )
        )

        oldSemester = userActiveSemester()

        setUserActiveSemester(model.id)

        try:
            if form.transfer_parts.data:
                model.transferPartsFrom(oldSemester)

            oldSemesterExperiments = [
                semesterExperiment.experiment for semesterExperiment in oldSemester.semester_experiments
            ]

            for experiment in Experiment.query.filter(Experiment.active == True):
                newSemesterExperiment = SemesterExperiment(experiment=experiment, semester=model)

                if form.transfer_assistants.data:
                    for oldSemesterExperiment in oldSemester.semester_experiments:
                        if oldSemesterExperiment.experiment == experiment:
                            newSemesterExperiment.assistants = oldSemesterExperiment.assistants

                self.session.add(newSemesterExperiment)

            self.session.commit()
        except Exception as ex:
            flash(str(ex), "error")

            self.session.rollback()


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

    def queryFilter(self):
        return 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()]},
    }

    column_extra_row_actions = [
        EndpointLinkRowAction(
            "glyphicon glyphicon-time",
            id_arg="flt1_0",
            title="Experiments history",
            endpoint="experimentmark.index_view",
        )
    ]


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


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


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

        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
        marks = range(16)[::-1]
        final_part_mark = SelectField("Final Part Mark", choices=[(-1, "-")] + list(zip(marks, marks)), coerce=int)

    form = EditForm

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

    partGroupPartMismatchException = "Student's part and group's part do not match!"

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

    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 update_model(self, form, model):
        if form.final_part_mark.data == -1:
            form.final_part_mark.data = None

        return super().update_model(form, model)


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(
            "Part", query_factory=partQueryFactory, validators=[DataRequired()], allow_blank=True, blank_text="-"
        )
        part_students = QuerySelectMultipleField("Part Students", query_factory=partStudentsQueryFactory)

    class EditForm(CreateForm):
        part = None

    form = EditForm

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

    partStudentPartPartMismatchException = "Group's part and student's part do not match!"

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

    def create_model(self, form):
        try:
            model = Group.customInit(form.part.data, form.part_students.data)

            self.session.add(model)
            self.on_model_change(form, model, True)
            self.session.commit()
        except Exception as ex:
            flash(str(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)


class ExperimentView(SecureModelView):
    can_view_details = True
    column_filters = ["active"]
    column_list = ["label", "title", "active"]
    column_details_list = column_list + [
        "desciption",
        "wiki_link",
        "room",
        "building",
        "responsibility",
        "duration_in_days",
        "oral_weighting",
        "protocol_weighting",
        "final_weighting",
        "semester_experiments",
    ]
    column_editable_list = ["active"]


class SemesterExperimentView(SecureModelView):
    column_list = ["experiment", "semester", "assistants"]

    def queryFilter(self):
        return SemesterExperiment.semester == userActiveSemester()


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


def assistantQueryFactory():
    return Assistant.query.filter(Assistant.user_id.in_([user.id for user in User.query.filter(User.active == True)]))


assistantBlankText = "Auto assign if experiment has only one assistant"


class GroupExperimentView(SecureModelView):
    class CreateForm(Form):
        def semesterExperimentQueryFactory():
            return SemesterExperiment.query.filter(SemesterExperiment.semester == userActiveSemester())

        group = QuerySelectField(
            "Group", query_factory=groupQueryFactory, validators=[DataRequired()], allow_blank=True, blank_text="-"
        )
        semester_experiment = QuerySelectField(
            "Semester Experiment",
            query_factory=semesterExperimentQueryFactory,
            validators=[DataRequired()],
            allow_blank=True,
            blank_text="-",
        )

        appointment1_date = DateField("Appointment-1 Date", validators=[Optional()])
        appointment1_special = BooleanField("Appointment-1 Special", default=False)
        appointment1_assistant = QuerySelectField(
            "Appointment-1 Assistant",
            query_factory=assistantQueryFactory,
            allow_blank=True,
            blank_text=assistantBlankText,
        )

        appointment2_date = DateField("Appointment-2 Date", validators=[Optional()])
        appointment2_special = BooleanField("Appointment-2 Special", default=False)
        appointment2_assistant = QuerySelectField(
            "Appointment-2 Assistant",
            query_factory=assistantQueryFactory,
            allow_blank=True,
            blank_text=assistantBlankText,
        )

    form = CreateForm

    can_edit = False
    column_list = ["group", "semester_experiment", "appointments", "experiment_marks"]
    column_filters = ["group", "semester_experiment.experiment", "appointments"]

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

    def create_model(self, form):
        try:
            model = GroupExperiment.customInit(semester_experiment=form.semester_experiment.data, group=form.group.data)

            self.session.add(model)

            for date, special, assistant in zip(
                [form.appointment1_date.data, form.appointment2_date.data],
                [form.appointment1_special.data, form.appointment2_special.data],
                [form.appointment1_assistant.data, form.appointment2_assistant.data],
            ):
                if date:
                    appointment = Appointment.customInit(date, special, assistant, model)

                    self.session.add(appointment)

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

            self.session.rollback()
        else:
            if model.appointments:
                flash(f"Appointments {model.appointments} added.", "success")

            self.after_model_change(form, model, True)
            return model


def groupExperimentQueryFactory():
    return GroupExperiment.query.filter(
        GroupExperiment.semester_experiment_id.in_(
            [
                semesterExperiment.id
                for semesterExperiment in SemesterExperiment.query.filter(
                    SemesterExperiment.semester == userActiveSemester()
                )
            ]
        )
    )


class AppointmentView(SecureModelView):
    class CreateForm(Form):
        group_experiment = QuerySelectField(
            "Group Experiment", query_factory=groupExperimentQueryFactory, validators=[DataRequired()]
        )
        date = DateField("Date", validators=[DataRequired()])
        special = BooleanField("Special", default=False)
        assistant = QuerySelectField(
            "Assistant",
            query_factory=assistantQueryFactory,
            allow_blank=True,
            blank_text=assistantBlankText,
        )

    form = CreateForm

    column_filters = ["date", "special", "group_experiment.group", "group_experiment.group.part", "assistant"]

    def queryFilter(self):
        return Appointment.group_experiment_id.in_(
            groupExperiment.id for groupExperiment in groupExperimentQueryFactory()
        )

    def create_model(self, form):
        try:
            model = Appointment.customInit(
                form.date.data, form.special.data, form.assistant.data, form.group_experiment.data
            )

            self.session.add(model)

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

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

    def update_model(self, form, model):
        try:
            model.customUpdate(form.date.data, form.special.data, form.assistant.data, form.group_experiment.data)

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

            self.session.rollback()

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


class ExperimentMarkView(SecureModelView):
    class StudentIdFilter(BaseSQLAFilter):
        def apply(self, query, value, alias=None):
            return query.filter(self.column == value)

        def operation(self):
            return "equals"

        def validate(self, value):
            if Student.query.get(value):
                return True
            else:
                return False

    class CreateForm(Form):
        part_student = QuerySelectField(
            "Part Student",
            query_factory=partStudentsQueryFactory,
            validators=[DataRequired()],
            allow_blank=True,
            blank_text="-",
        )
        group_experiment = QuerySelectField(
            "Group Experiment",
            query_factory=groupExperimentQueryFactory,
            validators=[DataRequired()],
            allow_blank=True,
            blank_text="-",
        )

        oral_mark = SelectField("Oral Mark", choices=markChoices, coerce=int)
        protocol_mark = SelectField("Protocol Mark", choices=markChoices, coerce=int)

    form = CreateForm

    column_filters = [
        StudentIdFilter(PartStudent.id, "Student / ID"),
        "part_student.student",
        "group_experiment.semester_experiment.semester",
        "group_experiment.semester_experiment.experiment",
        "assistant",
        "edited_by_admin",
    ]

    def queryFilter(self=None):
        return GroupExperiment.semester_experiment_id.in_(
            [
                semesterExperiment.id
                for semesterExperiment in SemesterExperiment.query.filter(
                    SemesterExperiment.semester == userActiveSemester()
                )
    ]
        )

    def checkForm(form):
        if form.oral_mark.data == -1:
            form.oral_mark.data = None

        if form.protocol_mark.data == -1:
            form.protocol_mark.data = None

        return form

    def create_model(self, form):
        form = ExperimentMarkView.checkForm(form)

        return super().create_model(form)

    def update_model(self, form, model):
        form = ExperimentMarkView.checkForm(form)

        return super().update_model(form, model)

    def after_model_change(self, form, model, is_created):
        if model.oral_mark or model.protocol_mark:
            try:
                model.edited_by_admin = True

                self.session.commit()
            except Exception as ex:
                flash(str(ex), "error")

                self.session.rollback()


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(AppointmentView(Appointment, db.session))
admin.add_view(ExperimentMarkView(ExperimentMark, db.session))
admin.add_view(ExperimentView(Experiment, db.session))
admin.add_view(SemesterExperimentView(SemesterExperiment, db.session))
admin.add_view(AssistantView(Assistant, db.session))
admin.add_view(PartView(Part, db.session))
admin.add_view(SemesterView(Semester, db.session))
admin.add_view(UserView(User, db.session))
admin.add_view(RoleView(Role, db.session))

with app.app_context():
    semesters = Semester.query.order_by(Semester.id)
    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")))