109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
import datetime
|
|
|
|
from apiflask import APIBlueprint
|
|
from flask import abort, Response, make_response
|
|
|
|
from ...base.utils import paginate_models
|
|
from ...dependencies import db
|
|
from ...examination_schedule.models import ExaminationSchedule, Enrollment
|
|
from ...students.models import Group
|
|
from ...project_supervisor.models import ProjectSupervisor
|
|
from ..schemas import ExaminationScheduleSchema, ExaminationScheduleUpdateSchema, MessageSchema, \
|
|
ExaminationSchedulesQuerySchema, ExaminationSchedulesPaginationSchema
|
|
from ..utils import generate_examination_schedule_pdf_file
|
|
|
|
bp = APIBlueprint("examination_schedule", __name__, url_prefix="/examination_schedule")
|
|
|
|
|
|
@bp.get('/')
|
|
@bp.input(ExaminationSchedulesQuerySchema, location='query')
|
|
@bp.output(ExaminationSchedulesPaginationSchema)
|
|
def list_examination_schedule(query: dict) -> dict:
|
|
page = query.get('page')
|
|
per_page = query.get('per_page')
|
|
data = paginate_models(page, ExaminationSchedule.query, per_page)
|
|
return {'examination_schedules': data['items'], 'max_pages': data['max_pages']}
|
|
|
|
|
|
@bp.post('/')
|
|
@bp.input(ExaminationScheduleSchema)
|
|
@bp.output(MessageSchema)
|
|
def create_examination_schedule(data: dict) -> dict:
|
|
examination_schedule = ExaminationSchedule(**data)
|
|
db.session.add(examination_schedule)
|
|
db.session.commit()
|
|
return {"message": "Examination schedule was created!"}
|
|
|
|
|
|
@bp.put('/<int:id>/')
|
|
@bp.input(ExaminationScheduleSchema)
|
|
@bp.output(MessageSchema)
|
|
def update_title_examination_schedule(id: int, data: dict) -> dict:
|
|
examination_schedule_query = db.session.query(ExaminationSchedule).filter(ExaminationSchedule.id == id)
|
|
examination_schedule = examination_schedule_query.first()
|
|
|
|
if examination_schedule is None:
|
|
abort(404, "Examination schedule doesn't exist!")
|
|
examination_schedule_query.update(data)
|
|
db.session.commit()
|
|
return {"message": "Examination schedule was updated!"}
|
|
|
|
|
|
@bp.delete('/<int:id>/')
|
|
@bp.output(MessageSchema)
|
|
def delete_examination_schedule(id: int) -> dict:
|
|
examination_schedule = db.session.query(ExaminationSchedule).filter(ExaminationSchedule.id == id).first()
|
|
if examination_schedule is None:
|
|
abort(404, "Examination schedule doesn't exist!")
|
|
db.session.delete(examination_schedule)
|
|
db.session.commit()
|
|
return {"message": "Examination schedule was deleted!"}
|
|
|
|
|
|
@bp.put('/<int:id>/date')
|
|
@bp.input(ExaminationScheduleUpdateSchema)
|
|
@bp.output(MessageSchema)
|
|
def set_date_of_examination_schedule(id: int, data: dict) -> dict:
|
|
examination_schedule_query = db.session.query(ExaminationSchedule).filter(ExaminationSchedule.id == id)
|
|
examination_schedule = examination_schedule_query.first()
|
|
|
|
if examination_schedule is None:
|
|
abort(404, "Examination schedule doesn't exist!")
|
|
|
|
if data['start_date'] > data['end_date']:
|
|
abort(400, "Invalid data! End date must be greater than start date!")
|
|
|
|
examination_schedule_query.update(data)
|
|
db.session.commit()
|
|
return {"message": "You set date of examination schedule!"}
|
|
|
|
|
|
@bp.post('/<int:examination_schedule_id>/download/')
|
|
def download_examination_schedule(examination_schedule_id: int) -> Response:
|
|
examination_schedule = db.session.query(ExaminationSchedule). \
|
|
filter(ExaminationSchedule.id == examination_schedule_id).first()
|
|
|
|
if examination_schedule is None:
|
|
abort(404, "Examination schedule doesn't exist!")
|
|
|
|
distinct_dates = db.session.query(db.func.Date(Enrollment.start_date)).distinct().all()
|
|
# print(distinct_dates)
|
|
nested_enrollments = []
|
|
for d in distinct_dates:
|
|
date_tmp = datetime.datetime.strptime(d[0], "%Y-%m-%d").date()
|
|
enrollment = db.session.query(Enrollment).join(ExaminationSchedule, isouter=True).\
|
|
join(Group, isouter=True).join(ProjectSupervisor, isouter=True). \
|
|
filter(ExaminationSchedule.id == examination_schedule_id).filter(
|
|
db.func.Date(Enrollment.start_date) == date_tmp).all()
|
|
nested_enrollments.append(enrollment)
|
|
|
|
# print(nested_enrollments)
|
|
pdf = generate_examination_schedule_pdf_file(examination_schedule.title, nested_enrollments)
|
|
title = examination_schedule.title.replace("-", "_").split()
|
|
filename = "_".join(title)
|
|
|
|
response = make_response(pdf)
|
|
response.headers['Content-Type'] = 'application/pdf'
|
|
response.headers['Content-Disposition'] = 'attachment; filename=%s.pdf' % filename
|
|
return response
|