from typing import Tuple from random import randint from flask import Blueprint, request, Response from marshmallow import ValidationError 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 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("//", 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("//", 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("//", 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: return {"error": "You didn't attach a csv file!"}, 400 try: students = parse_csv(uploaded_file) except InvalidNameOrTypeHeaderException: return {"error": "Invalid format of csv file!"}, 400 for student in students: db.session.add(student) db.session.commit() return {"message": "Students was created by uploading csv file!"}, 200