system-pri/backend/app/project_supervisor/routes/enrollments.py

200 lines
6.8 KiB
Python
Raw Normal View History

from apiflask import APIBlueprint
from flask import abort
from sqlalchemy import and_, or_
from ...base.mode import EnrollmentsMode
from ...base.schemas import MessageSchema
from ...dependencies import db
from ...examination_schedule.models import (
ExaminationSchedule,
TemporaryAvailability,
TermOfDefence,
)
from ..models import ProjectSupervisor
from ..schemas import (
ListOfFreeTimesSchema,
ListOfTermOfDefenceSchema,
TemporaryProjectSupervisorSchema,
TimeAvailabilityCreateSchema,
)
bp = APIBlueprint("enrollments", __name__, url_prefix="/")
@bp.post("/<int:examination_schedule_id>/enrollments/")
@bp.input(TimeAvailabilityCreateSchema)
@bp.output(MessageSchema)
def set_your_free_time_to_examination_schedule(
examination_schedule_id: int, data: dict
) -> dict:
# this code will be removed
project_supervisor = ProjectSupervisor.query.filter(
ProjectSupervisor.id == data["project_supervisor_id"]
).first()
if project_supervisor is None:
abort(404, "ProjectSupervisor doesn't exist!")
print(project_supervisor)
################
examination_schedule = ExaminationSchedule.query.filter(
ExaminationSchedule.id == examination_schedule_id
).first()
if examination_schedule is None:
abort(404, "Examination schedule doesn't exist!")
if examination_schedule.year_group_id != project_supervisor.year_group_id:
abort(400, "You are not assigned to this year group!")
if examination_schedule.open_enrollments != EnrollmentsMode.INIT.value:
abort(400, "Enrollments has started or closed! You have been delayed!")
sd = data.get("start_date")
ed = data.get("end_date")
if sd > ed:
abort(400, "Invalid data! End date must be greater than start date!")
start_date = examination_schedule.start_date
end_date = examination_schedule.end_date
if not (
start_date.timestamp() <= sd.timestamp()
and end_date.timestamp() >= ed.timestamp()
):
abort(400, "Invalid date range!")
ta = (
TemporaryAvailability.query.filter(
TemporaryAvailability.examination_schedule_id == examination_schedule_id
)
.filter(TemporaryAvailability.project_supervisor_id == project_supervisor.id)
.filter(
or_(
and_(
TemporaryAvailability.start_date >= sd,
TemporaryAvailability.start_date < ed,
TemporaryAvailability.end_date >= ed,
),
and_(
TemporaryAvailability.start_date <= sd,
TemporaryAvailability.end_date > sd,
TemporaryAvailability.end_date <= ed,
),
)
)
.first()
)
if ta is not None:
abort(
400,
"Invalid date ranges. You set your free time "
"in this date range! Choose another date!",
)
ta = TemporaryAvailability(**data, examination_schedule_id=examination_schedule_id)
db.session.add(ta)
db.session.commit()
return {"message": "You have just assigned your free time!"}
@bp.delete(
"/<int:examination_schedule_id>/enrollments/<int:temporary_availability_id>/"
)
@bp.input(TemporaryProjectSupervisorSchema)
@bp.output(MessageSchema)
def delete_your_free_time_from_examination_schedule(
examination_schedule_id: int, temporary_availability_id: int, data: dict
) -> dict:
# this code will be removed
project_supervisor = (
db.session.query(ProjectSupervisor)
.filter(ProjectSupervisor.id == data["id"])
.first()
)
if project_supervisor is None:
abort(404, "ProjectSupervisor doesn't exist!")
################
examination_schedule = ExaminationSchedule.query.filter(
ExaminationSchedule.id == examination_schedule_id
).first()
if examination_schedule is None:
abort(404, "Examination schedule doesn't exist!")
if examination_schedule.open_enrollments != EnrollmentsMode.INIT.value:
abort(400, "Enrollments has started or closed! You have been delayed!")
ta = TemporaryAvailability.query.filter(
TemporaryAvailability.examination_schedule_id == examination_schedule_id,
TemporaryAvailability.project_supervisor_id == project_supervisor.id,
TemporaryAvailability.id == temporary_availability_id,
).first()
if ta is None:
abort(404, "Your free time doesn't exist!")
db.session.delete(ta)
db.session.commit()
return {"message": "You have just removed your free time!"}
@bp.get("/<int:examination_schedule_id>/temporary-availabilities/")
@bp.input(TemporaryProjectSupervisorSchema, location="query")
@bp.output(ListOfFreeTimesSchema)
def list_enrollments_for_project_supervisor(
examination_schedule_id: int, data: dict
) -> dict:
# this code will be removed
project_supervisor = (
db.session.query(ProjectSupervisor)
.filter(ProjectSupervisor.id == data["id"])
.first()
)
if project_supervisor is None:
abort(404, "Not found project supervisor!")
################
examination_schedule = ExaminationSchedule.query.filter(
ExaminationSchedule.id == examination_schedule_id
).first()
if examination_schedule is None:
abort(404, "Not found examination schedule!")
if examination_schedule.open_enrollments != EnrollmentsMode.INIT.value:
abort(400, "Enrollments has started or closed! You have been delayed!")
# list of your term of defences first enrollment
ta = TemporaryAvailability.query.filter(
TemporaryAvailability.examination_schedule_id == examination_schedule_id,
TemporaryAvailability.project_supervisor_id == project_supervisor.id,
).all()
return {"free_times": ta}
@bp.get("/<int:examination_schedule_id>/term-of-defences/")
@bp.input(TemporaryProjectSupervisorSchema, location="query")
@bp.output(ListOfTermOfDefenceSchema)
def list_created_term_of_defences_by_coordinator_for_project_supervisor(
examination_schedule_id: int, data: dict
) -> dict:
# this code will be removed
project_supervisor = (
db.session.query(ProjectSupervisor)
.filter(ProjectSupervisor.id == data["id"])
.first()
)
if project_supervisor is None:
abort(404, "Not found project supervisor!")
################
es = ExaminationSchedule.query.filter(
ExaminationSchedule.id == examination_schedule_id,
ExaminationSchedule.year_group_id,
).first()
if es is None:
abort(404, "Not found examination schedule!")
# list of your free times first enrollment
td = (
TermOfDefence.query.join(TermOfDefence.members_of_committee)
.filter(TermOfDefence.examination_schedule_id == examination_schedule_id)
.filter_by(id=project_supervisor.id)
.all()
)
return {"term_of_defences": td}