131 lines
4.7 KiB
Python
131 lines
4.7 KiB
Python
from typing import Tuple
|
|
from random import randint
|
|
from itertools import islice
|
|
|
|
from flask import Blueprint, request, Response
|
|
from marshmallow import ValidationError
|
|
from sqlalchemy.exc import IntegrityError
|
|
from flask_sqlalchemy import get_debug_queries
|
|
|
|
from ...students.models import Student
|
|
from ..schemas import StudentSchema, StudentEditSchema, StudentCreateSchema
|
|
from ...dependencies import db
|
|
from ..utils import parse_csv
|
|
from ..exceptions import InvalidNameOrTypeHeaderException
|
|
from ...base.utils import paginate_models, is_allowed_extensions
|
|
|
|
bp = Blueprint("students", __name__, url_prefix="/students")
|
|
|
|
|
|
@bp.route("/", methods=["GET"])
|
|
def list_students() -> Tuple[dict, int]:
|
|
students_schema = StudentSchema(many=True)
|
|
|
|
fullname = request.args.get('fullname')
|
|
order_by_first_name = request.args.get('order_by_first_name')
|
|
order_by_last_name = request.args.get('order_by_last_name')
|
|
page = request.args.get('page')
|
|
|
|
student_query = Student.search_by_fullname_and_order_by_first_name_or_last_name(fullname, order_by_first_name,
|
|
order_by_last_name)
|
|
|
|
response = paginate_models(page, student_query)
|
|
if isinstance(response, tuple):
|
|
return response
|
|
# print(get_debug_queries()[0])
|
|
return {"students": students_schema.dump(response['items']), "max_pages": response['max_pages']}, 200
|
|
|
|
|
|
@bp.route("/<int:index>/", methods=["GET"])
|
|
def detail_student(index: int) -> Tuple[dict, int]:
|
|
student_schema = StudentSchema()
|
|
student = Student.query.filter_by(index=index).first()
|
|
if student is not None:
|
|
return student_schema.dump(student), 200
|
|
return {"error": f"Student with {index} index doesn't exist!"}, 404
|
|
|
|
|
|
@bp.route("/<int:index>/", methods=["DELETE"])
|
|
def delete_student(index: int) -> Tuple[dict, int]:
|
|
student = Student.query.filter_by(index=index).first()
|
|
if student is not None:
|
|
db.session.delete(student)
|
|
db.session.commit()
|
|
return {"message": "Student was deleted!"}, 202
|
|
return {"error": f"Student with {index} index doesn't exist!"}, 404
|
|
|
|
|
|
@bp.route("/<int:index>/", methods=["PUT"])
|
|
def edit_student(index: int) -> Tuple[dict, int]:
|
|
request_data = request.get_json()
|
|
student_schema = StudentEditSchema()
|
|
|
|
try:
|
|
data = student_schema.load(request_data)
|
|
if not data:
|
|
return {"error": "You have passed empty data!"}, 400
|
|
|
|
student_query = Student.query.filter_by(index=index)
|
|
student = student_query.first()
|
|
if student is None:
|
|
return {"error": "Not Found student!"}, 404
|
|
|
|
student_query.update(data)
|
|
db.session.commit()
|
|
except ValidationError as e:
|
|
return {"error": e.messages}, 422
|
|
return {"message": "Student was updated!"}, 200
|
|
|
|
|
|
@bp.route("/", methods=["POST"])
|
|
def create_student() -> Tuple[dict, int] or Response:
|
|
request_data = request.get_json()
|
|
student_schema = StudentCreateSchema()
|
|
|
|
try:
|
|
data = student_schema.load(request_data)
|
|
|
|
index = data['index']
|
|
student = Student.query.filter_by(index=index).first()
|
|
if student is not None:
|
|
return {"error": "Student has already exists!"}, 400
|
|
|
|
dummy_email = f'student{randint(1, 300_000)}@gmail.com'
|
|
student = Student(**data, email=dummy_email)
|
|
db.session.add(student)
|
|
db.session.commit()
|
|
except ValidationError as e:
|
|
return {'error': e.messages}, 422
|
|
return {"message": "Student was created!"}, 200
|
|
|
|
|
|
@bp.route("/upload/", methods=["POST"])
|
|
def upload_students() -> Tuple[dict, int]:
|
|
"""Maybe in the future move to celery workers"""
|
|
if (uploaded_file := request.files.get('file')) is None or uploaded_file.filename == '':
|
|
return {"error": "You didn't attach a csv file!"}, 400
|
|
|
|
if uploaded_file and is_allowed_extensions(uploaded_file.filename):
|
|
try:
|
|
students = parse_csv(uploaded_file)
|
|
except InvalidNameOrTypeHeaderException:
|
|
return {"error": "Invalid format of csv file!"}, 400
|
|
|
|
try:
|
|
while True:
|
|
sliced_students = islice(students, 5)
|
|
list_of_students = list(sliced_students)
|
|
|
|
if len(list_of_students) == 0:
|
|
break
|
|
db.session.add_all(list_of_students)
|
|
db.session.commit()
|
|
except IntegrityError as e:
|
|
# print(e)
|
|
# in the future create sql query checks index and add only these students, which didn't exist in db
|
|
return {"error": "These students have already exist!"}, 400
|
|
else:
|
|
return {"error": "Invalid extension of file"}, 400
|
|
|
|
return {"message": "Students was created by uploading csv file!"}, 200
|