236 lines
6.5 KiB
Python
236 lines
6.5 KiB
Python
from functools import wraps
|
|
from flask_login import login_required, login_user, logout_user
|
|
from app.bp import bp
|
|
from flask import jsonify, request
|
|
from app.errors import error_response
|
|
from flask_login import current_user
|
|
|
|
from app import login, db
|
|
from app.models import Course, User, Assignment
|
|
|
|
|
|
@login.user_loader
|
|
def load_user(user_id):
|
|
return User.query.get(user_id)
|
|
|
|
|
|
def check_data(data, required_fields):
|
|
for f in required_fields:
|
|
if f not in data:
|
|
return f
|
|
return None
|
|
|
|
|
|
def instructor_required(func):
|
|
@wraps(func)
|
|
def dec(*args, **kwargs):
|
|
if current_user.role != "instructor":
|
|
return error_response(400, "User is not instructor!")
|
|
return func(*args, **kwargs)
|
|
|
|
return dec
|
|
|
|
|
|
@bp.route("/login", methods=["POST"])
|
|
def login_route():
|
|
data = request.get_json()
|
|
|
|
if current_user.is_authenticated:
|
|
return error_response(400, "A user is already logged in!")
|
|
|
|
if not data.get("username") or not data.get("password"):
|
|
return error_response(400, "Must supply username and password")
|
|
|
|
user = User.query.filter_by(username=data["username"]).first()
|
|
if not user:
|
|
return error_response(400, "User not found")
|
|
|
|
if not user.check_password(data.get("password")):
|
|
return error_response(400, "Invalid password")
|
|
|
|
login_user(user)
|
|
resp = jsonify(user.to_dict())
|
|
return resp
|
|
|
|
|
|
@bp.route("/logout", methods=["POST"])
|
|
@login_required
|
|
def logout_route():
|
|
resp = jsonify(current_user.to_dict())
|
|
logout_user()
|
|
return resp
|
|
|
|
|
|
@bp.route("/register", methods=["POST"])
|
|
def register():
|
|
data = request.get_json()
|
|
required_fields = ["role", "username", "email", "password", "password2"]
|
|
if f := check_data(data, required_fields):
|
|
return error_response(400, f"Must supply {f}")
|
|
|
|
if User.query.filter_by(username=data["username"]).first():
|
|
return error_response(409, "User with that username already exists")
|
|
|
|
if User.query.filter_by(email=data["email"]).first():
|
|
return error_response(409, "User with that email already exists")
|
|
|
|
u = User()
|
|
u.from_dict(data, new_user=True)
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
|
|
resp = jsonify(u.to_dict())
|
|
return resp
|
|
|
|
|
|
@bp.route("/course", methods=["POST"])
|
|
@login_required
|
|
def create_course():
|
|
data = request.get_json()
|
|
|
|
required_fields = ["name", "course_code", "description", "instructor"]
|
|
|
|
if f := check_data(data, required_fields):
|
|
return error_response(400, f"Must supply {f}")
|
|
|
|
u = User.query.get(data["instructor"])
|
|
if not u:
|
|
return error_response(400, f"User with id {data['instructor']} does not exist")
|
|
|
|
c = Course.query.filter_by(course_code=data["course_code"]).first()
|
|
if c:
|
|
return error_response(
|
|
400, f"Course with course code {data['course_code']} already exists"
|
|
)
|
|
|
|
if u.role != "instructor":
|
|
return error_response(400, "User is not instructor")
|
|
|
|
c = Course()
|
|
c.from_dict(data)
|
|
u.enroll(c)
|
|
db.session.add(c)
|
|
db.session.commit()
|
|
|
|
return jsonify(c.to_dict())
|
|
|
|
|
|
@bp.route("/user/<int:id>/courses", methods=["GET"])
|
|
@login_required
|
|
def get_courses(id):
|
|
u = User.query.get(id)
|
|
d = {"courses": []}
|
|
for c in u.enrolled_courses.all():
|
|
d["courses"].append(c.to_dict())
|
|
resp = jsonify(d)
|
|
return resp
|
|
|
|
|
|
@bp.route("/course/<int:id>/students", methods=["GET"])
|
|
@login_required
|
|
def get_students_in_course(id):
|
|
c = Course.query.get(id)
|
|
if not c:
|
|
return error_response(400, f"course with id {id} not found")
|
|
|
|
students = c.students.filter_by(role="student")
|
|
resp = {"students": []}
|
|
for s in students:
|
|
resp["students"].append(s.to_dict())
|
|
return jsonify(resp)
|
|
|
|
@bp.route("/course/<int:id>/assignments", methods=["GET"])
|
|
@login_required
|
|
def get_assignments_in_course(id):
|
|
c = Course.query.get(id)
|
|
if not c:
|
|
return error_response(400, f"course with id {id} not found")
|
|
|
|
assignments = c.assignments.all()
|
|
resp = {"assignments": []}
|
|
for a in assignments:
|
|
resp["assignments"].append(a.to_dict())
|
|
return jsonify(resp)
|
|
|
|
|
|
@bp.route("/course/<int:id>", methods=["GET"])
|
|
@login_required
|
|
def get_course(id):
|
|
c = Course.query.get(id)
|
|
if not c:
|
|
return error_response(400, f"course with id {id} not found")
|
|
resp = jsonify(c.to_dict())
|
|
return resp
|
|
|
|
|
|
@bp.route("/user/<int:uid>/enroll/<int:cid>", methods=["POST", "DELETE"])
|
|
@login_required
|
|
@instructor_required
|
|
def enroll_student(uid, cid):
|
|
u = User.query.get(uid)
|
|
if not u:
|
|
return error_response(400, f"User with id {uid} does not exist")
|
|
|
|
c = Course.query.get(cid)
|
|
if not c:
|
|
return error_response(400, f"Course with id {cid} does not exist")
|
|
|
|
if request.method == "POST":
|
|
if not u.enroll(c):
|
|
return error_response(
|
|
400, f"User {uid} is already enrolled in course {cid}"
|
|
)
|
|
|
|
elif request.method == "DELETE":
|
|
if not u.unenroll(c):
|
|
return error_response(400, f"User {uid} is not enrolled in course {cid}")
|
|
|
|
resp = {"user": u.to_dict(), "course": c.to_dict()}
|
|
return jsonify(resp)
|
|
|
|
|
|
@bp.route("/user/<string:username>/enroll/<int:cid>", methods=["POST"])
|
|
@login_required
|
|
@instructor_required
|
|
def enroll_student_by_username(username, cid):
|
|
u = User.query.filter_by(username=username).first()
|
|
if not u:
|
|
return error_response(400, f"User with username {username} does not exist")
|
|
|
|
c = Course.query.get(cid)
|
|
if not c:
|
|
return error_response(400, f"Course with id {cid} does not exist")
|
|
|
|
if request.method == "POST":
|
|
if not u.enroll(c):
|
|
return error_response(
|
|
400, f"User {u.id} is already enrolled in course {cid}"
|
|
)
|
|
|
|
elif request.method == "DELETE":
|
|
if not u.unenroll(c):
|
|
return error_response(400, f"User {u.id} is not enrolled in course {cid}")
|
|
|
|
resp = {"user": u.to_dict(), "course": c.to_dict()}
|
|
return jsonify(resp)
|
|
|
|
|
|
@bp.route("/assignment", methods=["POST"])
|
|
@login_required
|
|
@instructor_required
|
|
def create_assignment():
|
|
data = request.get_json()
|
|
required_fields = ["name", "description", "course_id", "due_date"]
|
|
if f := check_data(data, required_fields):
|
|
return error_response(400, f"Must supply {f}")
|
|
|
|
c = Course.query.get(data["course_id"])
|
|
if not c:
|
|
return error_response(400, f"Course with id {data['course_id']} does not exist")
|
|
|
|
a = Assignment()
|
|
a.from_dict(data)
|
|
db.session.add(a)
|
|
db.session.commit()
|
|
return jsonify(a.to_dict())
|