101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
from typing import Tuple
|
|
from random import randint
|
|
|
|
from flask import Blueprint, request, make_response, jsonify, Response
|
|
from marshmallow import ValidationError
|
|
|
|
from ...students.models import Student
|
|
from ..schemas import StudentSchema, StudentEditSchema, StudentCreateSchema
|
|
from ...dependencies import db
|
|
from ..utils import parse_csv
|
|
from ..exceptions import InvalidNameOrTypeHeaderException
|
|
|
|
bp = Blueprint("students", __name__, url_prefix="/students")
|
|
|
|
|
|
@bp.route("/", methods=["GET"])
|
|
def list_students() -> Tuple[dict, int]:
|
|
students_schema = StudentSchema(many=True)
|
|
students = Student.query.all()
|
|
return {"students": students_schema.dump(students)}, 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:
|
|
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
|