diff --git a/backend/project/__init__.py b/backend/project/__init__.py index 33450700..b0c21275 100644 --- a/backend/project/__init__.py +++ b/backend/project/__init__.py @@ -6,8 +6,7 @@ from .db_in import db from .endpoints.index.index import index_bp from .endpoints.projects.project_endpoint import project_bp -from .endpoints.courses import courses_bp - +from .endpoints.courses.courses_config import courses_bp def create_app(): diff --git a/backend/project/__main__.py b/backend/project/__main__.py index 2f312c85..a4bd122b 100644 --- a/backend/project/__main__.py +++ b/backend/project/__main__.py @@ -1,11 +1,10 @@ """Main entry point for the application.""" -from sys import path +from dotenv import load_dotenv from project import create_app_with_db from project.db_in import url -path.append(".") - if __name__ == "__main__": + load_dotenv() app = create_app_with_db(url) app.run(debug=True) diff --git a/backend/project/db_in.py b/backend/project/db_in.py index ebcc02dd..57a572fa 100644 --- a/backend/project/db_in.py +++ b/backend/project/db_in.py @@ -2,13 +2,10 @@ import os from flask_sqlalchemy import SQLAlchemy -from dotenv import load_dotenv from sqlalchemy import URL db = SQLAlchemy() -load_dotenv() - DATABSE_NAME = os.getenv("POSTGRES_DB") DATABASE_USER = os.getenv("POSTGRES_USER") DATABASE_PASSWORD = os.getenv("POSTGRES_PASSWORD") diff --git a/backend/project/endpoints/courses.py b/backend/project/endpoints/courses.py deleted file mode 100644 index 4a4def72..00000000 --- a/backend/project/endpoints/courses.py +++ /dev/null @@ -1,623 +0,0 @@ -"""Course api point""" - -from os import getenv -import dataclasses -from typing import List -from dotenv import load_dotenv -from flask import Blueprint, jsonify, request -from flask import abort -from flask_restful import Api, Resource -from sqlalchemy.exc import SQLAlchemyError -from project.models.course_relations import CourseAdmin, CourseStudent -from project.models.users import User -from project.models.courses import Course -from project.models.projects import Project -from project import db - -courses_bp = Blueprint("courses", __name__) -courses_api = Api(courses_bp) - -load_dotenv() -API_URL = getenv("API_HOST") - - -def execute_query_abort_if_db_error(query, url, query_all=False): - """ - Execute the given SQLAlchemy query and handle any SQLAlchemyError that might occur. - If query_all == True, the query will be executed with the all() method, - otherwise with the first() method. - Args: - query (Query): The SQLAlchemy query to execute. - - Returns: - ResultProxy: The result of the query if successful, otherwise aborts with error 500. - """ - try: - if query_all: - result = query.all() - else: - result = query.first() - except SQLAlchemyError as e: - response = json_message(str(e)) - response["url"] = url - abort(500, description=response) - return result - - -def add_abort_if_error(to_add, url): - """ - Add a new object to the database - and handle any SQLAlchemyError that might occur. - - Args: - to_add (object): The object to add to the database. - """ - try: - db.session.add(to_add) - except SQLAlchemyError as e: - db.session.rollback() - response = json_message(str(e)) - response["url"] = url - abort(500, description=response) - - -def delete_abort_if_error(to_delete, url): - """ - Deletes the given object from the database - and aborts the request with a 500 error if a SQLAlchemyError occurs. - - Args: - - to_delete: The object to be deleted from the database. - """ - try: - db.session.delete(to_delete) - except SQLAlchemyError as e: - db.session.rollback() - response = json_message(str(e)) - response["url"] = url - abort(500, description=response) - - -def commit_abort_if_error(url): - """ - Commit the current session and handle any SQLAlchemyError that might occur. - """ - try: - db.session.commit() - except SQLAlchemyError as e: - db.session.rollback() - response = json_message(str(e)) - response["url"] = url - abort(500, description=response) - - -def abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant): - """ - Check if the current user is authorized to appoint new admins to a course. - - Args: - course_id (int): The ID of the course. - - Raises: - HTTPException: If the current user is not authorized or - if the UID of the person to be made an admin is missing in the request body. - """ - url = API_URL + "/courses/" + str(course_id) + "/admins" - abort_if_uid_is_none(teacher, url) - - course = get_course_abort_if_not_found(course_id) - - if teacher != course.teacher: - response = json_message("Only the teacher of a course can appoint new admins") - response["url"] = url - abort(403, description=response) - - if not assistant: - response = json_message( - "uid of person to make admin is required in the request body" - ) - response["url"] = url - abort(400, description=response) - - -def abort_if_none_uid_student_uids_or_non_existant_course_id( - course_id, uid, student_uids -): - """ - Check the request to assign new students to a course. - - Args: - course_id (int): The ID of the course. - - Raises: - 403: If the user is not authorized to assign new students to the course. - 400: If the request body does not contain the required 'students' field. - """ - url = API_URL + "/courses/" + str(course_id) + "/students" - get_course_abort_if_not_found(course_id) - abort_if_no_user_found_for_uid(uid, url) - query = CourseAdmin.query.filter_by(uid=uid, course_id=course_id) - admin_relation = execute_query_abort_if_db_error(query, url) - if not admin_relation: - message = "Not authorized to assign new students to course with id " + str( - course_id - ) - response = json_message(message) - response["url"] = url - abort(403, description=response) - - if not student_uids: - message = """To assign new students to a course, - you should have a students field with a list of uids in the request body""" - response = json_message(message) - response["url"] = url - abort(400, description=response) - - -def abort_if_uid_is_none(uid, url): - """ - Check whether the uid is None if so - abort with error 400 - """ - if uid is None: - response = json_message("There should be a uid in the request query") - response["url"] = url - abort(400, description=response) - - -def abort_if_no_user_found_for_uid(uid, url): - """ - Check if a user exists based on the provided uid. - - Args: - uid (int): The unique identifier of the user. - - Raises: - NotFound: If the user with the given uid is not found. - """ - query = User.query.filter_by(uid=uid) - user = execute_query_abort_if_db_error(query, url) - - if not user: - response = json_message("User with uid " + uid + " was not found") - response["url"] = url - abort(404, description=response) - return user - - -def get_admin_relation(uid, course_id): - """ - Retrieve the CourseAdmin object for the given uid and course. - - Args: - uid (int): The user ID. - course_id (int): The course ID. - - Returns: - CourseAdmin: The CourseAdmin object if the user is an admin, otherwise None. - """ - return execute_query_abort_if_db_error( - CourseAdmin.query.filter_by(uid=uid, course_id=course_id), - url=API_URL + "/courses/" + str(course_id) + "/admins", - ) - - -def json_message(message): - """ - Create a json message with the given message. - - Args: - message (str): The message to include in the json. - - Returns: - dict: The message in a json format. - """ - return {"message": message} - - -def get_course_abort_if_not_found(course_id): - """ - Get a course by its ID. - - Args: - course_id (int): The course ID. - - Returns: - Course: The course with the given ID. - """ - query = Course.query.filter_by(course_id=course_id) - course = execute_query_abort_if_db_error(query, API_URL + "/courses") - - if not course: - response = json_message("Course not found") - response["url"] = API_URL + "/courses" - abort(404, description=response) - - return course - - -class CourseForUser(Resource): - """Api endpoint for the /courses link""" - - def get(self): - """ " - Get function for /courses this will be the main endpoint - to get all courses and filter by given query parameter like /courses?parameter=... - parameters can be either one of the following: teacher,ufora_id,name. - """ - query = Course.query - if "teacher" in request.args: - query = query.filter_by(course_id=request.args.get("teacher")) - if "ufora_id" in request.args: - query = query.filter_by(ufora_id=request.args.get("ufora_id")) - if "name" in request.args: - query = query.filter_by(name=request.args.get("name")) - results:List[Course] = execute_query_abort_if_db_error( - query, url=API_URL + "/courses", query_all=True - ) - courses = [ - {**dataclasses.asdict(course), - "url":f"{API_URL}/courses/{course.course_id}"} - for course in results - ] - message = "Succesfully retrieved all courses with given parameters" - response = json_message(message) - response["data"] = courses - response["url"] = API_URL + "/courses" - return jsonify(response) - - def post(self): - """ - This function will create a new course - if the body of the post contains a name and uid is an admin or teacher - """ - abort_url = API_URL + "/courses" - uid = request.args.get("uid") - abort_if_uid_is_none(uid, abort_url) - - user = abort_if_no_user_found_for_uid(uid, abort_url) - - if not user.is_teacher: - message = ( - "Only teachers or admins can create new courses, you are unauthorized" - ) - return json_message(message), 403 - - data = request.get_json() - - if "name" not in data: - message = "Missing 'name' in the request body" - return json_message(message), 400 - - name = data["name"] - new_course = Course(name=name, teacher=uid) - if "ufora_id" in data: - new_course.ufora_id = data["ufora_id"] - - add_abort_if_error(new_course, abort_url) - commit_abort_if_error(abort_url) - - admin_course = CourseAdmin(uid=uid, course_id=new_course.course_id) - add_abort_if_error(admin_course, abort_url) - commit_abort_if_error(abort_url) - - message = (f"Course with name: {name} and" - f"course_id:{new_course.course_id} was succesfully created") - response = json_message(message) - data = { - "course_id": API_URL + "/courses/" + str(new_course.course_id), - "name": new_course.name, - "teacher": API_URL + "/users/" + new_course.teacher, - "ufora_id": new_course.ufora_id if new_course.ufora_id else "None", - } - response["data"] = data - response["url"] = API_URL + "/courses/" + str(new_course.course_id) - return response, 201 - - -class CourseByCourseId(Resource): - """Api endpoint for the /courses/course_id link""" - - def get(self, course_id): - """ - This get function will return all the related projects of the course - in the following form: - { - course: course with course_id - projects: [ - list of all projects that have course_id - where projects are jsons containing the title, deadline and project_id - ] - } - """ - abort_url = API_URL + "/courses" - uid = request.args.get("uid") - abort_if_uid_is_none(uid, abort_url) - admin = get_admin_relation(uid, course_id) - query = CourseStudent.query.filter_by(uid=uid, course_id=course_id) - student = execute_query_abort_if_db_error(query, abort_url) - - if not (admin or student): - message = "User is not an admin, nor a student of this course" - return json_message(message), 404 - - course = get_course_abort_if_not_found(course_id) - query = Project.query.filter_by(course_id=course_id) - abort_url = API_URL + "/courses/" + str(course_id) - # course does exist so url should be to the id - project_uids = [ - API_URL + "/projects/" + project.project_id - for project in execute_query_abort_if_db_error( - query, abort_url, query_all=True - ) - ] - query = CourseAdmin.query.filter_by(course_id=course_id) - admin_uids = [ - API_URL + "/users/" + admin.uid - for admin in execute_query_abort_if_db_error( - query, abort_url, query_all=True - ) - ] - query = CourseStudent.query.filter_by(course_id=course_id) - student_uids = [ - API_URL + "/users/" + student.uid - for student in execute_query_abort_if_db_error( - query, abort_url, query_all=True - ) - ] - - data = { - "ufora_id": course.ufora_id, - "teacher": API_URL + "/users/" + course.teacher, - "admins": admin_uids, - "students": student_uids, - "projects": project_uids, - } - response = json_message( - "Succesfully retrieved course with course_id: " + str(course_id) - ) - response["data"] = data - response["url"] = API_URL + "/courses/" + str(course_id) - return response - - def delete(self, course_id): - """ - This function will delete the course with course_id - """ - abort_url = API_URL + "/courses/" - uid = request.args.get("uid") - abort_if_uid_is_none(uid, abort_url) - - admin = get_admin_relation(uid, course_id) - - if not admin: - message = "You are not an admin of this course and so you cannot delete it" - return json_message(message), 403 - - course = get_course_abort_if_not_found(course_id) - abort_url = API_URL + "/courses/" + str(course_id) - # course does exist so url should be to the id - delete_abort_if_error(course, abort_url) - commit_abort_if_error(abort_url) - - response = { - "message": "Succesfully deleted course with course_id: " + str(course_id), - "url": API_URL + "/courses", - } - return response - - def patch(self, course_id): - """ - This function will update the course with course_id - """ - abort_url = API_URL + "/courses/" - uid = request.args.get("uid") - abort_if_uid_is_none(uid, abort_url) - - admin = get_admin_relation(uid, course_id) - - if not admin: - message = "You are not an admin of this course and so you cannot update it" - return json_message(message), 403 - - data = request.get_json() - course = get_course_abort_if_not_found(course_id) - abort_url = API_URL + "/courses/" + str(course_id) - if "name" in data: - course.name = data["name"] - if "teacher" in data: - course.teacher = data["teacher"] - if "ufora_id" in data: - course.ufora_id = data["ufora_id"] - - commit_abort_if_error(abort_url) - response = json_message( - "Succesfully updated course with course_id: " + str(course_id) - ) - response["url"] = API_URL + "/courses/" + str(course_id) - data = { - "course_id": API_URL + "/courses/" + str(course.course_id), - "name": course.name, - "teacher": API_URL + "/users/" + course.teacher, - "ufora_id": course.ufora_id if course.ufora_id else "None", - } - response["data"] = data - return response, 200 - - -class CourseForAdmins(Resource): - """ - This class will handle post and delete queries to - the /courses/course_id/admins url, only the teacher of a course can do this - """ - - def get(self, course_id): - """ - This function will return all the admins of a course - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/admins" - get_course_abort_if_not_found(course_id) - - query = CourseAdmin.query.filter_by(course_id=course_id) - admin_uids = [ - API_URL + "/users/" + a.uid - for a in execute_query_abort_if_db_error(query, abort_url, query_all=True) - ] - response = json_message( - "Succesfully retrieved all admins of course " + str(course_id) - ) - response["data"] = admin_uids - response["url"] = abort_url # not actually aborting here tho heheh - return jsonify(admin_uids) - - def post(self, course_id): - """ - Api endpoint for adding new admins to a course, can only be done by the teacher - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/admins" - teacher = request.args.get("uid") - data = request.get_json() - assistant = data.get("admin_uid") - abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant) - - query = User.query.filter_by(uid=assistant) - new_admin = execute_query_abort_if_db_error(query, abort_url) - if not new_admin: - message = ( - "User to make admin was not found, please request with a valid uid" - ) - return json_message(message), 404 - - admin_relation = CourseAdmin(uid=assistant, course_id=course_id) - add_abort_if_error(admin_relation, abort_url) - commit_abort_if_error(abort_url) - response = json_message( - f"Admin assistant added to course {course_id}" - ) - response["url"] = abort_url - data = { - "course_id": API_URL + "/courses/" + str(course_id), - "uid": API_URL + "/users/" + assistant, - } - response["data"] = data - return response, 201 - - def delete(self, course_id): - """ - Api endpoint for removing admins of a course, can only be done by the teacher - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/admins" - teacher = request.args.get("uid") - data = request.get_json() - assistant = data.get("admin_uid") - abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant) - - query = CourseAdmin.query.filter_by(uid=assistant, course_id=course_id) - admin_relation = execute_query_abort_if_db_error(query, abort_url) - if not admin_relation: - message = "Course with given admin not found" - return json_message(message), 404 - - delete_abort_if_error(admin_relation, abort_url) - commit_abort_if_error(abort_url) - - message = ( - f"Admin {assistant}" - f" was succesfully removed from course {course_id}" - ) - response = json_message(message) - response["url"] = abort_url - return response, 204 - - -class CourseToAddStudents(Resource): - """ - Class that will respond to the /courses/course_id/students link - teachers should be able to assign and remove students from courses, - and everyone should be able to list all students assigned to a course - """ - - def get(self, course_id): - """ - Get function at /courses/course_id/students - to get all the users assigned to a course - everyone can get this data so no need to have uid query in the link - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/students" - get_course_abort_if_not_found(course_id) - - query = CourseStudent.query.filter_by(course_id=course_id) - student_uids = [ - API_URL + "/users/" + s.uid - for s in execute_query_abort_if_db_error(query, abort_url, query_all=True) - ] - response = json_message( - "Succesfully retrieved all students of course " + str(course_id) - ) - response["data"] = student_uids - response["url"] = abort_url - return response - - def post(self, course_id): - """ - Allows admins of a course to assign new students by posting to: - /courses/course_id/students with a list of uid in the request body under key "students" - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/students" - uid = request.args.get("uid") - data = request.get_json() - student_uids = data.get("students") - abort_if_none_uid_student_uids_or_non_existant_course_id( - course_id, uid, student_uids - ) - - for uid in student_uids: - query = CourseStudent.query.filter_by(uid=uid, course_id=course_id) - student_relation = execute_query_abort_if_db_error(query, abort_url) - if student_relation: - db.session.rollback() - message = ( - "Student with uid " + uid + " is already assigned to the course" - ) - return json_message(message), 400 - add_abort_if_error(CourseStudent(uid=uid, course_id=course_id), abort_url) - commit_abort_if_error(abort_url) - response = json_message("User were succesfully added to the course") - response["url"] = abort_url - data = {"students": [API_URL + "/users/" + uid for uid in student_uids]} - response["data"] = data - return response, 201 - - def delete(self, course_id): - """ - This function allows admins of a course to remove students by sending a delete request to - /courses/course_id/students with inside the request body - a field "students" = [list of uids to unassign] - """ - abort_url = API_URL + "/courses/" + str(course_id) + "/students" - uid = request.args.get("uid") - data = request.get_json() - student_uids = data.get("students") - abort_if_none_uid_student_uids_or_non_existant_course_id( - course_id, uid, student_uids - ) - - for uid in student_uids: - query = CourseStudent.query.filter_by(uid=uid, course_id=course_id) - student_relation = execute_query_abort_if_db_error(query, abort_url) - if student_relation: - delete_abort_if_error(student_relation, abort_url) - commit_abort_if_error(abort_url) - - response = json_message("User were succesfully removed from the course") - response["url"] = API_URL + "/courses/" + str(course_id) + "/students" - return response - - -courses_api.add_resource(CourseForUser, "/courses") - -courses_api.add_resource(CourseByCourseId, "/courses/") - -courses_api.add_resource(CourseForAdmins, "/courses//admins") - -courses_api.add_resource(CourseToAddStudents, "/courses//students") diff --git a/backend/project/endpoints/courses/course_admin_relation.py b/backend/project/endpoints/courses/course_admin_relation.py new file mode 100644 index 00000000..c4793a21 --- /dev/null +++ b/backend/project/endpoints/courses/course_admin_relation.py @@ -0,0 +1,100 @@ +""" +This module will handle the /courses//admins endpoint +It will allow the teacher of a course to add and remove admins from a course +""" + +from os import getenv +from urllib.parse import urljoin +from dotenv import load_dotenv + +from flask import request +from flask_restful import Resource + +from project.models.course_relations import CourseAdmin +from project.models.users import User +from project.endpoints.courses.courses_utils import ( + execute_query_abort_if_db_error, + commit_abort_if_error, + delete_abort_if_error, + get_course_abort_if_not_found, + abort_if_not_teacher_or_none_assistant, + json_message +) +from project.utils.query_agent import query_selected_from_model, insert_into_model + +load_dotenv() +API_URL = getenv("API_HOST") +RESPONSE_URL = urljoin(API_URL + "/", "courses") + +class CourseForAdmins(Resource): + """ + This class will handle post and delete queries to + the /courses/course_id/admins url, only the teacher of a course can do this + """ + + def get(self, course_id): + """ + This function will return all the admins of a course + """ + abort_url = urljoin(f"{RESPONSE_URL}/" , f"{str(course_id)}/", "admins") + get_course_abort_if_not_found(course_id) + + return query_selected_from_model( + CourseAdmin, + abort_url, + select_values=["uid"], + url_mapper={"uid": urljoin(f"{API_URL}/", "users")}, + filters={"course_id": course_id}, + ) + + def post(self, course_id): + """ + Api endpoint for adding new admins to a course, can only be done by the teacher + """ + abort_url = urljoin(f"{RESPONSE_URL}/" , f"{str(course_id)}/", "admins") + teacher = request.args.get("uid") + data = request.get_json() + assistant = data.get("admin_uid") + abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant) + + query = User.query.filter_by(uid=assistant) + new_admin = execute_query_abort_if_db_error(query, abort_url) + if not new_admin: + message = ( + "User to make admin was not found, please request with a valid uid" + ) + return json_message(message), 404 + + return insert_into_model( + CourseAdmin, + {"uid": assistant, "course_id": course_id}, + abort_url, + "uid" + ) + + def delete(self, course_id): + """ + Api endpoint for removing admins of a course, can only be done by the teacher + """ + abort_url = urljoin(f"{RESPONSE_URL}/" , f"{str(course_id)}/", "admins") + teacher = request.args.get("uid") + data = request.get_json() + assistant = data.get("admin_uid") + abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant) + + query = CourseAdmin.query.filter_by(uid=assistant, course_id=course_id) + admin_relation = execute_query_abort_if_db_error(query, abort_url) + if not admin_relation: + message = "Course with given admin not found" + return json_message(message), 404 + + delete_abort_if_error(admin_relation, abort_url) + commit_abort_if_error(abort_url) + + message = ( + f"Admin {assistant}" + f" was succesfully removed from course {course_id}" + ) + response = json_message(message) + response["url"] = abort_url + return response, 204 diff --git a/backend/project/endpoints/courses/course_details.py b/backend/project/endpoints/courses/course_details.py new file mode 100644 index 00000000..605aae01 --- /dev/null +++ b/backend/project/endpoints/courses/course_details.py @@ -0,0 +1,111 @@ +""" +This file contains the api endpoint for the /courses/course_id url +This file is responsible for handling the requests made to the /courses/course_id url +and returning the appropriate response as well as handling the requests made to the +/courses/course_id/admins and /courses/course_id/students urls +""" + +from os import getenv +from urllib.parse import urljoin + +from dotenv import load_dotenv + +from flask import request +from flask_restful import Resource +from sqlalchemy.exc import SQLAlchemyError + +from project.models.courses import Course +from project.models.course_relations import CourseAdmin, CourseStudent + +from project import db +from project.utils.query_agent import delete_by_id_from_model, patch_by_id_from_model + +load_dotenv() +API_URL = getenv("API_HOST") +RESPONSE_URL = urljoin(API_URL + "/", "courses") + +class CourseByCourseId(Resource): + """Api endpoint for the /courses/course_id link""" + + def get(self, course_id): + """ + This get function will return all the related projects of the course + in the following form: + { + course: course with course_id + projects: [ + list of all projects that have course_id + where projects are jsons containing the title, deadline and project_id + ] + } + """ + try: + course_details = db.session.query( + Course.course_id, + Course.name, + Course.ufora_id, + Course.teacher + ).filter( + Course.course_id == course_id).first() + + if not course_details: + return { + "message": "Course not found", + "url": RESPONSE_URL + }, 404 + + admins = db.session.query(CourseAdmin.uid).filter( + CourseAdmin.course_id == course_id + ).all() + + students = db.session.query(CourseStudent.uid).filter( + CourseStudent.course_id == course_id + ).all() + + user_url = urljoin(API_URL + "/", "users") + + admin_ids = [ urljoin(f"{user_url}/" , admin[0]) for admin in admins] + student_ids = [ urljoin(f"{user_url}/", student[0]) for student in students] + + result = { + 'course_id': course_details.course_id, + 'name': course_details.name, + 'ufora_id': course_details.ufora_id, + 'teacher': course_details.teacher, + 'admins': admin_ids, + 'students': student_ids + } + + return { + "message": f"Succesfully retrieved course with course_id: {str(course_id)}", + "data": result, + "url": urljoin(f"{RESPONSE_URL}/", str(course_id)) + } + except SQLAlchemyError: + return { + "error": "Something went wrong while querying the database.", + "url": RESPONSE_URL}, 500 + + def delete(self, course_id): + """ + This function will delete the course with course_id + """ + return delete_by_id_from_model( + Course, + "course_id", + course_id, + RESPONSE_URL + ) + + def patch(self, course_id): + """ + This function will update the course with course_id + """ + + return patch_by_id_from_model( + Course, + "course_id", + course_id, + RESPONSE_URL, + request.json + ) diff --git a/backend/project/endpoints/courses/course_student_relation.py b/backend/project/endpoints/courses/course_student_relation.py new file mode 100644 index 00000000..4a5a6a55 --- /dev/null +++ b/backend/project/endpoints/courses/course_student_relation.py @@ -0,0 +1,111 @@ +""" +This file contains the class CourseToAddStudents which is a +resource for the /courses/course_id/students link. +This class will allow admins of a course to assign and remove students from courses, +and everyone should be able to list all students assigned to a course. +""" + +from os import getenv +from urllib.parse import urljoin + +from dotenv import load_dotenv + +from flask import request +from flask_restful import Resource + +from project import db +from project.models.course_relations import CourseStudent +from project.endpoints.courses.courses_utils import ( + execute_query_abort_if_db_error, + add_abort_if_error, + commit_abort_if_error, + delete_abort_if_error, + get_course_abort_if_not_found, + abort_if_none_uid_student_uids_or_non_existant_course_id, + json_message, +) + +from project.utils.query_agent import query_selected_from_model + +load_dotenv() +API_URL = getenv("API_HOST") +RESPONSE_URL = urljoin(API_URL + "/", "courses") + +class CourseToAddStudents(Resource): + """ + Class that will respond to the /courses/course_id/students link + teachers should be able to assign and remove students from courses, + and everyone should be able to list all students assigned to a course + """ + + def get(self, course_id): + """ + Get function at /courses/course_id/students + to get all the users assigned to a course + everyone can get this data so no need to have uid query in the link + """ + abort_url = f"{API_URL}/courses/{str(course_id)}/students" + get_course_abort_if_not_found(course_id) + + return query_selected_from_model( + CourseStudent, + abort_url, + select_values=["uid"], + url_mapper={"uid": urljoin(f"{API_URL}/", "users")}, + filters={"course_id": course_id} + ) + + def post(self, course_id): + """ + Allows admins of a course to assign new students by posting to: + /courses/course_id/students with a list of uid in the request body under key "students" + """ + abort_url = f"{API_URL}/courses/{str(course_id)}/students" + uid = request.args.get("uid") + data = request.get_json() + student_uids = data.get("students") + abort_if_none_uid_student_uids_or_non_existant_course_id( + course_id, uid, student_uids + ) + + for uid in student_uids: + query = CourseStudent.query.filter_by(uid=uid, course_id=course_id) + student_relation = execute_query_abort_if_db_error(query, abort_url) + if student_relation: + db.session.rollback() + message = ( + f"Student with uid {uid} is already assigned to the course" + ) + return json_message(message), 400 + add_abort_if_error(CourseStudent(uid=uid, course_id=course_id), abort_url) + commit_abort_if_error(abort_url) + response = json_message("Users were succesfully added to the course") + response["url"] = abort_url + data = {"students": [f"{API_URL}/users/{uid}" for uid in student_uids]} + response["data"] = data + return response, 201 + + def delete(self, course_id): + """ + This function allows admins of a course to remove students by sending a delete request to + /courses/course_id/students with inside the request body + a field "students" = [list of uids to unassign] + """ + abort_url = f"{API_URL}/courses/{str(course_id)}/students" + uid = request.args.get("uid") + data = request.get_json() + student_uids = data.get("students") + abort_if_none_uid_student_uids_or_non_existant_course_id( + course_id, uid, student_uids + ) + + for uid in student_uids: + query = CourseStudent.query.filter_by(uid=uid, course_id=course_id) + student_relation = execute_query_abort_if_db_error(query, abort_url) + if student_relation: + delete_abort_if_error(student_relation, abort_url) + commit_abort_if_error(abort_url) + + response = json_message("Users were succesfully removed from the course") + response["url"] = f"{API_URL}/courses/{str(course_id)}/students" + return response diff --git a/backend/project/endpoints/courses/courses.py b/backend/project/endpoints/courses/courses.py new file mode 100644 index 00000000..c06d7dfc --- /dev/null +++ b/backend/project/endpoints/courses/courses.py @@ -0,0 +1,51 @@ +""" +This file contains the main endpoint for the /courses url. +This endpoint is used to get all courses and filter by given +query parameter like /courses?parameter=... +parameters can be either one of the following: teacher,ufora_id,name. +""" + +from os import getenv +from urllib.parse import urljoin +from dotenv import load_dotenv + +from flask import request +from flask_restful import Resource + +from project.models.courses import Course +from project.utils.query_agent import query_selected_from_model, insert_into_model + +load_dotenv() +API_URL = getenv("API_HOST") +RESPONSE_URL = urljoin(API_URL + "/", "courses") + +class CourseForUser(Resource): + """Api endpoint for the /courses link""" + + def get(self): + """ " + Get function for /courses this will be the main endpoint + to get all courses and filter by given query parameter like /courses?parameter=... + parameters can be either one of the following: teacher,ufora_id,name. + """ + + return query_selected_from_model( + Course, + RESPONSE_URL, + url_mapper={"course_id": RESPONSE_URL}, + filters=request.args + ) + + def post(self): + """ + This function will create a new course + if the body of the post contains a name and uid is an admin or teacher + """ + + return insert_into_model( + Course, + request.json, + RESPONSE_URL, + "course_id", + required_fields=["name", "teacher"] + ) diff --git a/backend/project/endpoints/courses/courses_config.py b/backend/project/endpoints/courses/courses_config.py new file mode 100644 index 00000000..f791031f --- /dev/null +++ b/backend/project/endpoints/courses/courses_config.py @@ -0,0 +1,32 @@ +""" +This file is used to configure the courses blueprint and the courses api. +It is used to define the routes for the courses blueprint and the +corresponding api endpoints. + +The courses blueprint is used to define the routes for the courses api +endpoints and the courses api is used to define the routes for the courses +api endpoints. +""" + +from flask import Blueprint +from flask_restful import Api + +from project.endpoints.courses.courses import CourseForUser +from project.endpoints.courses.course_details import CourseByCourseId +from project.endpoints.courses.course_admin_relation import CourseForAdmins +from project.endpoints.courses.course_student_relation import CourseToAddStudents + +courses_bp = Blueprint("courses", __name__) +courses_api = Api(courses_bp) + +courses_bp.add_url_rule("/courses", + view_func=CourseForUser.as_view('course_endpoint')) + +courses_bp.add_url_rule("/courses/", + view_func=CourseByCourseId.as_view('course_by_course_id')) + +courses_bp.add_url_rule("/courses//admins", + view_func=CourseForAdmins.as_view('course_admins')) + +courses_bp.add_url_rule("/courses//students", + view_func=CourseToAddStudents.as_view('course_students')) diff --git a/backend/project/endpoints/courses/courses_utils.py b/backend/project/endpoints/courses/courses_utils.py new file mode 100644 index 00000000..747bc3c2 --- /dev/null +++ b/backend/project/endpoints/courses/courses_utils.py @@ -0,0 +1,234 @@ +""" +This module contains utility functions for the courses endpoints. +The functions are used to interact with the database and handle errors. +""" + +from os import getenv +from urllib.parse import urljoin + +from dotenv import load_dotenv +from flask import abort +from sqlalchemy.exc import SQLAlchemyError + +from project import db +from project.models.course_relations import CourseAdmin +from project.models.users import User +from project.models.courses import Course + +load_dotenv() +API_URL = getenv("API_HOST") +RESPONSE_URL = urljoin(API_URL + "/", "courses") + +def execute_query_abort_if_db_error(query, url, query_all=False): + """ + Execute the given SQLAlchemy query and handle any SQLAlchemyError that might occur. + If query_all == True, the query will be executed with the all() method, + otherwise with the first() method. + Args: + query (Query): The SQLAlchemy query to execute. + + Returns: + ResultProxy: The result of the query if successful, otherwise aborts with error 500. + """ + try: + if query_all: + result = query.all() + else: + result = query.first() + except SQLAlchemyError as e: + response = json_message(str(e)) + response["url"] = url + abort(500, description=response) + return result + + +def add_abort_if_error(to_add, url): + """ + Add a new object to the database + and handle any SQLAlchemyError that might occur. + + Args: + to_add (object): The object to add to the database. + """ + try: + db.session.add(to_add) + except SQLAlchemyError as e: + db.session.rollback() + response = json_message(str(e)) + response["url"] = url + abort(500, description=response) + + +def delete_abort_if_error(to_delete, url): + """ + Deletes the given object from the database + and aborts the request with a 500 error if a SQLAlchemyError occurs. + + Args: + - to_delete: The object to be deleted from the database. + """ + try: + db.session.delete(to_delete) + except SQLAlchemyError as e: + db.session.rollback() + response = json_message(str(e)) + response["url"] = url + abort(500, description=response) + + +def commit_abort_if_error(url): + """ + Commit the current session and handle any SQLAlchemyError that might occur. + """ + try: + db.session.commit() + except SQLAlchemyError as e: + db.session.rollback() + response = json_message(str(e)) + response["url"] = url + abort(500, description=response) + + +def abort_if_not_teacher_or_none_assistant(course_id, teacher, assistant): + """ + Check if the current user is authorized to appoint new admins to a course. + + Args: + course_id (int): The ID of the course. + + Raises: + HTTPException: If the current user is not authorized or + if the UID of the person to be made an admin is missing in the request body. + """ + url = f"{API_URL}/courses/{str(course_id)}/admins" + abort_if_uid_is_none(teacher, url) + + course = get_course_abort_if_not_found(course_id) + + if teacher != course.teacher: + response = json_message("Only the teacher of a course can appoint new admins") + response["url"] = url + abort(403, description=response) + + if not assistant: + response = json_message( + "uid of person to make admin is required in the request body" + ) + response["url"] = url + abort(400, description=response) + + +def abort_if_none_uid_student_uids_or_non_existant_course_id( + course_id, uid, student_uids +): + """ + Check the request to assign new students to a course. + + Args: + course_id (int): The ID of the course. + + Raises: + 403: If the user is not authorized to assign new students to the course. + 400: If the request body does not contain the required 'students' field. + """ + url = f"{API_URL}/courses/{str(course_id)}/students" + get_course_abort_if_not_found(course_id) + abort_if_no_user_found_for_uid(uid, url) + query = CourseAdmin.query.filter_by(uid=uid, course_id=course_id) + admin_relation = execute_query_abort_if_db_error(query, url) + if not admin_relation: + message = "Not authorized to assign new students to course with id " + str( + course_id + ) + response = json_message(message) + response["url"] = url + abort(403, description=response) + + if not student_uids: + message = """To assign new students to a course, + you should have a students field with a list of uids in the request body""" + response = json_message(message) + response["url"] = url + abort(400, description=response) + + +def abort_if_uid_is_none(uid, url): + """ + Check whether the uid is None if so + abort with error 400 + """ + if uid is None: + response = json_message("There should be a uid in the request query") + response["url"] = url + abort(400, description=response) + + +def abort_if_no_user_found_for_uid(uid, url): + """ + Check if a user exists based on the provided uid. + + Args: + uid (int): The unique identifier of the user. + + Raises: + NotFound: If the user with the given uid is not found. + """ + query = User.query.filter_by(uid=uid) + user = execute_query_abort_if_db_error(query, url) + + if not user: + response = json_message(f"User with uid {uid} was not found") + response["url"] = url + abort(404, description=response) + return user + + +def get_admin_relation(uid, course_id): + """ + Retrieve the CourseAdmin object for the given uid and course. + + Args: + uid (int): The user ID. + course_id (int): The course ID. + + Returns: + CourseAdmin: The CourseAdmin object if the user is an admin, otherwise None. + """ + return execute_query_abort_if_db_error( + CourseAdmin.query.filter_by(uid=uid, course_id=course_id), + url=f"{API_URL}/courses/{str(course_id)}/admins", + ) + + +def json_message(message): + """ + Create a json message with the given message. + + Args: + message (str): The message to include in the json. + + Returns: + dict: The message in a json format. + """ + return {"message": message} + + +def get_course_abort_if_not_found(course_id): + """ + Get a course by its ID. + + Args: + course_id (int): The course ID. + + Returns: + Course: The course with the given ID. + """ + query = Course.query.filter_by(course_id=course_id) + course = execute_query_abort_if_db_error(query, f"{API_URL}/courses") + + if not course: + response = json_message("Course not found") + response["url"] = f"{API_URL}/courses" + abort(404, description=response) + + return course diff --git a/backend/project/endpoints/projects/project_detail.py b/backend/project/endpoints/projects/project_detail.py index 88989247..e2314bd9 100644 --- a/backend/project/endpoints/projects/project_detail.py +++ b/backend/project/endpoints/projects/project_detail.py @@ -4,18 +4,18 @@ the corresponding project is 1 """ from os import getenv -from dotenv import load_dotenv +from urllib.parse import urljoin -from flask import jsonify -from flask_restful import Resource, abort -from sqlalchemy import exc -from project.endpoints.projects.endpoint_parser import parse_project_params +from flask import request +from flask_restful import Resource -from project import db from project.models.projects import Project +from project.utils.query_agent import query_by_id_from_model, delete_by_id_from_model, \ + patch_by_id_from_model + -load_dotenv() API_URL = getenv('API_HOST') +RESPONSE_URL = urljoin(API_URL, "projects") class ProjectDetail(Resource): """ @@ -24,14 +24,6 @@ class ProjectDetail(Resource): for implementing get, delete and put methods """ - def abort_if_not_present(self, project): - """ - Check if the project exists in the database - and if not abort the request and give back a 404 not found - """ - if project is None: - abort(404) - def get(self, project_id): """ Get method for listing a specific project @@ -39,22 +31,11 @@ def get(self, project_id): the id fetched from the url with the reaparse """ - try: - # fetch the project with the id that is specified in the url - project = Project.query.filter_by(project_id=project_id).first() - self.abort_if_not_present(project) - - # return the fetched project and return 200 OK status - return { - "data": jsonify(project).json, - "url": f"{API_URL}/projects/{project_id}", - "message": "Got project successfully" - }, 200 - except exc.SQLAlchemyError: - return { - "message": "Internal server error", - "url": f"{API_URL}/projects/{project_id}" - }, 500 + return query_by_id_from_model( + Project, + "project_id", + project_id, + RESPONSE_URL) def patch(self, project_id): """ @@ -62,30 +43,13 @@ def patch(self, project_id): filtered by id of that specific project """ - # get the project that need to be edited - project = Project.query.filter_by(project_id=project_id).first() - - # check which values are not None in the dict - # if it is not None it needs to be modified in the database - - # commit the changes and return the 200 OK code if it succeeds, else 500 - try: - var_dict = parse_project_params() - for key, value in var_dict.items(): - setattr(project, key, value) - db.session.commit() - # get the updated version - return { - "message": f"Succesfully changed project with id: {id}", - "url": f"{API_URL}/projects/{id}", - "data": project - }, 200 - except exc.SQLAlchemyError: - db.session.rollback() - return { - "message": f"Something unexpected happenend when trying to edit project {id}", - "url": f"{API_URL}/projects/{id}" - }, 500 + return patch_by_id_from_model( + Project, + "project_id", + project_id, + RESPONSE_URL, + request.json + ) def delete(self, project_id): """ @@ -93,25 +57,8 @@ def delete(self, project_id): done by project id """ - # fetch the project that needs to be removed - deleted_project = Project.query.filter_by(project_id=project_id).first() - - # check if its an existing one - self.abort_if_not_present(deleted_project) - - # if it exists delete it and commit the changes in the database - try: - db.session.delete(deleted_project) - db.session.commit() - - # return 200 if content is deleted succesfully - return { - "message": f"Project with id: {id} deleted successfully", - "url": f"{API_URL}/projects/{id} deleted successfully!", - "data": deleted_project - }, 200 - except exc.SQLAlchemyError: - return { - "message": f"Something unexpected happened when removing project {project_id}", - "url": f"{API_URL}/projects/{id}" - }, 500 + return delete_by_id_from_model( + Project, + "project_id", + project_id, + RESPONSE_URL) diff --git a/backend/project/endpoints/projects/projects.py b/backend/project/endpoints/projects/projects.py index f444e283..0834988f 100644 --- a/backend/project/endpoints/projects/projects.py +++ b/backend/project/endpoints/projects/projects.py @@ -2,18 +2,14 @@ Module that implements the /projects endpoint of the API """ from os import getenv -from dotenv import load_dotenv +from urllib.parse import urljoin -from flask import jsonify +from flask import request from flask_restful import Resource -from sqlalchemy import exc - -from project import db from project.models.projects import Project -from project.endpoints.projects.endpoint_parser import parse_project_params +from project.utils.query_agent import query_selected_from_model, insert_into_model -load_dotenv() API_URL = getenv('API_HOST') class ProjectsEndpoint(Resource): @@ -28,67 +24,23 @@ def get(self): Get method for listing all available projects that are currently in the API """ - try: - projects = Project.query.with_entities( - Project.project_id, - Project.title, - Project.descriptions - ).all() - - results = [{ - "project_id": row[0], - "title": row[1], - "descriptions": row[2] - } for row in projects] - # return all valid entries for a project and return a 200 OK code - return { - "data": results, - "url": f"{API_URL}/projects", - "message": "Projects fetched successfully" - }, 200 - except exc.SQLAlchemyError: - return { - "message": "Something unexpected happenend when trying to get the projects", - "url": f"{API_URL}/projects" - }, 500 + response_url = urljoin(API_URL, "projects") + return query_selected_from_model( + Project, + response_url, + select_values=["project_id", "title", "descriptions"], + url_mapper={"project_id": response_url}, + filters=request.args + ) def post(self): """ Post functionality for project using flask_restfull parse lib """ - args = parse_project_params() - - # create a new project object to add in the API later - new_project = Project( - title=args['title'], - descriptions=args['descriptions'], - assignment_file=args['assignment_file'], - deadline=args['deadline'], - course_id=args['course_id'], - visible_for_students=args['visible_for_students'], - archieved=args['archieved'], - test_path=args['test_path'], - script_name=args['script_name'], - regex_expressions=args['regex_expressions'] - ) - - # add the new project to the database and commit the changes - - try: - db.session.add(new_project) - db.session.commit() - new_project_json = jsonify(new_project).json - return { - "url": f"{API_URL}/projects/{new_project_json['project_id']}", - "message": "Project posted successfully", - "data": new_project_json - }, 201 - except exc.SQLAlchemyError: - return ({ - "url": f"{API_URL}/projects", - "message": "Something unexpected happenend when trying to add a new project", - "data": jsonify(new_project).json - }, 500) + return insert_into_model( + Project,request.json, + urljoin(API_URL, "/projects"), + "project_id") diff --git a/backend/project/models/courses.py b/backend/project/models/courses.py index 052e5d5f..8d3f0651 100644 --- a/backend/project/models/courses.py +++ b/backend/project/models/courses.py @@ -1,9 +1,9 @@ """The Course model""" + from dataclasses import dataclass from sqlalchemy import Integer, Column, ForeignKey, String from project import db - @dataclass class Course(db.Model): """This class described the courses table, diff --git a/backend/project/sessionmaker.py b/backend/project/sessionmaker.py index 9fbf1cad..0ab68f8e 100644 --- a/backend/project/sessionmaker.py +++ b/backend/project/sessionmaker.py @@ -1,11 +1,8 @@ """initialise a datab session""" from os import getenv -from dotenv import load_dotenv from sqlalchemy import create_engine, URL from sqlalchemy.orm import sessionmaker -load_dotenv() - url = URL.create( drivername="postgresql", username=getenv("POSTGRES_USER"), diff --git a/backend/project/utils/misc.py b/backend/project/utils/misc.py new file mode 100644 index 00000000..2995f8de --- /dev/null +++ b/backend/project/utils/misc.py @@ -0,0 +1,76 @@ +""" +This module contains miscellaneous utility functions. +These functions apply to a variety of use cases and are not specific to any one module. +""" + +from typing import Dict, List +from urllib.parse import urljoin +from sqlalchemy.ext.declarative import DeclarativeMeta + + +def map_keys_to_url(url_mapper: Dict[str, str], data: Dict[str, str]) -> Dict[str, str]: + """ + Maps keys to a url using a url mapper. + + Args: + url_mapper: Dict[str, str] - A dictionary that maps keys to urls. + data: Dict[str, str] - The data to map to urls. + + Returns: + A dictionary with the keys mapped to the urls. + """ + for key, value in data.items(): + if key in url_mapper: + data[key] = urljoin(url_mapper[key] + "/", str(value)) + return data + +def map_all_keys_to_url(url_mapper: Dict[str, str], data: List[Dict[str, str]]): + """ + Maps all keys to a url using a url mapper. + + Args: + url_mapper: Dict[str, str] - A dictionary that maps keys to urls. + data: List[Dict[str, str]] - The data to map to urls. + + Returns: + A list of dictionaries with the keys mapped to the urls. + """ + return [map_keys_to_url(url_mapper, entry) for entry in data] + +def model_to_dict(instance: DeclarativeMeta) -> Dict[str, str]: + """ + Converts an sqlalchemy model to a dictionary. + + Args: + instance: DeclarativeMeta - The instance of the model to convert to a dictionary. + + Returns: + A dictionary with the keys and values of the model. + """ + return {column.key: getattr(instance, column.key) for column in instance.__table__.columns} + +def models_to_dict(instances: List[DeclarativeMeta]) -> List[Dict[str, str]]: + """ + Converts a list of sqlalchemy models to a list of dictionaries. + + Args: + instances: List[DeclarativeMeta] - The instances of the models to convert to dictionaries. + + Returns: + A list of dictionaries with the keys and values of the models. + """ + return [model_to_dict(instance) for instance in instances] + + +def filter_model_fields(model: DeclarativeMeta, data: Dict[str, str]): + """ + Filters the data to only contain the fields of the model. + + Args: + model: DeclarativeMeta - The model to filter the data with. + data: Dict[str, str] - The data to filter. + + Returns: + A dictionary with the fields of the model. + """ + return {key: value for key, value in data.items() if hasattr(model, key)} diff --git a/backend/project/utils/query_agent.py b/backend/project/utils/query_agent.py new file mode 100644 index 00000000..8a688163 --- /dev/null +++ b/backend/project/utils/query_agent.py @@ -0,0 +1,210 @@ +""" +This module contains the functions to interact with the database. It contains functions to +delete, insert and query entries from the database. The functions are used by the routes +to interact with the database. +""" + +from typing import Dict, List, Union +from urllib.parse import urljoin +from flask import jsonify +from sqlalchemy import and_ +from sqlalchemy.ext.declarative import DeclarativeMeta +from sqlalchemy.orm.query import Query +from sqlalchemy.exc import SQLAlchemyError +from project.db_in import db +from project.utils.misc import map_all_keys_to_url, models_to_dict, filter_model_fields + +def delete_by_id_from_model( + model: DeclarativeMeta, + column_name: str, + column_id: int, + base_url: str): + """ + Deletes an entry from the database giving the model corresponding to a certain table, + a column name and its value. + + Args: + model: DeclarativeMeta - The model corresponding to the table to delete from. + column_name: str - The name of the column to delete from. + id: int - The id of the entry to delete. + + Returns: + A message indicating that the resource was deleted successfully if the operation was + successful, otherwise a message indicating that something went wrong while deleting from + the database. + """ + try: + result: DeclarativeMeta = model.query.filter( + getattr(model, column_name) == column_id + ).first() + + if not result: + return { + "message": "Resource not found", + "url": base_url}, 404 + db.session.delete(result) + db.session.commit() + return {"message": "Resource deleted successfully", + "url": base_url}, 200 + except SQLAlchemyError: + return {"error": "Something went wrong while deleting from the database.", + "url": base_url}, 500 + +def insert_into_model(model: DeclarativeMeta, + data: Dict[str, Union[str, int]], + response_url_base: str, + url_id_field: str, + required_fields: List[str] = None): + """ + Inserts a new entry into the database giving the model corresponding to a certain table + and the data to insert. + + Args: + model: DeclarativeMeta - The model corresponding to the table to insert into. + data: Dict[str, Union[str, int]] - The data to insert into the table. + response_url_base: str - The base url to use in the response. + + Returns: + The new entry inserted into the database if the operation was successful, otherwise + a message indicating that something went wrong while inserting into the database. + """ + try: + if required_fields is None: + required_fields = [] + # Check if all non-nullable fields are present in the data + missing_fields = [field for field in required_fields if field not in data] + + if missing_fields: + return {"error": f"Missing required fields: {', '.join(missing_fields)}", + "url": response_url_base}, 400 + + filtered_data = filter_model_fields(model, data) + new_instance: DeclarativeMeta = model(**filtered_data) + db.session.add(new_instance) + db.session.commit() + return jsonify({ + "data": new_instance, + "message": "Object created succesfully.", + "url": urljoin( + f"{response_url_base}/", + str(getattr(new_instance, url_id_field)))}), 201 + except SQLAlchemyError: + return jsonify({"error": "Something went wrong while inserting into the database.", + "url": response_url_base}), 500 + +def query_selected_from_model(model: DeclarativeMeta, + response_url: str, + url_mapper: Dict[str, str] = None, + select_values: List[str] = None, + filters: Dict[str, Union[str, int]]=None): + """ + Query entries from the database giving the model corresponding to a certain table and + the filters to apply to the query. + + + Args: + model: DeclarativeMeta - The model corresponding to the table to query from. + response_url: str - The base url to use in the response. + url_mapper: Dict[str, str] - A dictionary to map the keys of the response to urls. + select_values: List[str] - The columns to select from the table. + filters: Dict[str, Union[str, int]] - The filters to apply to the query. + + Returns: + The entries queried from the database if they exist, otherwise a message indicating + that the resource was not found. + """ + try: + query: Query = model.query + if filters: + filtered_filters = filter_model_fields(model, filters) + conditions: List[bool] = [] + for key, value in filtered_filters.items(): + conditions.append(getattr(model, key) == value) + query = query.filter(and_(*conditions)) + + if select_values: + query = query.with_entities(*[getattr(model, value) for value in select_values]) + query_result = query.all() + results = [] + for instance in query_result: + selected_instance = {} + for value in select_values: + selected_instance[value] = getattr(instance, value) + results.append(selected_instance) + else: + results = models_to_dict(query.all()) + if url_mapper: + results = map_all_keys_to_url(url_mapper, results) + response = {"data": results, + "message": "Resources fetched successfully", + "url": response_url} + return jsonify(response), 200 + except SQLAlchemyError: + return {"error": "Something went wrong while querying the database.", + "url": response_url}, 500 + +def query_by_id_from_model(model: DeclarativeMeta, + column_name: str, + column_id: int, + base_url: str): + """ + Query an entry from the database giving the model corresponding to a certain table, + a column name and its value. + + Args: + model: DeclarativeMeta - The model corresponding to the table to query from. + column_name: str - The name of the column to query from. + id: int - The id of the entry to query. + not_found_message: str - The message to return if the entry is not found. + + Returns: + The entry queried from the database if it exists, otherwise a message indicating + that the resource was not found. + + """ + try: + result: Query = model.query.filter(getattr(model, column_name) == column_id).first() + if not result: + return {"message": "Resource not found", "url": base_url}, 404 + return jsonify({ + "data": result, + "message": "Resource fetched correctly", + "url": urljoin(f"{base_url}/", str(column_id))}), 200 + except SQLAlchemyError: + return { + "error": "Something went wrong while querying the database.", + "url": base_url}, 500 + +def patch_by_id_from_model(model: DeclarativeMeta, + column_name: str, + column_id: int, + base_url: str, + data: Dict[str, Union[str, int]]): + """ + Update an entry from the database giving the model corresponding to a certain table, + a column name and its value. + + Args: + model: DeclarativeMeta - The model corresponding to the table to update. + column_name: str - The name of the column to update. + id: int - The id of the entry to update. + data: Dict[str, Union[str, int]] - The data to update the entry with. + + Returns: + The entry updated from the database if the operation was successful, otherwise + a message indicating that something went wrong while updating the entry. + """ + try: + result: Query = model.query.filter(getattr(model, column_name) == column_id).first() + if not result: + return {"message": "Resource not found", "url": base_url}, 404 + for key, value in data.items(): + setattr(result, key, value) + db.session.commit() + return jsonify({ + "data": result, + "message": "Resource updated successfully", + "url": urljoin(f"{base_url}/", str(column_id))}), 200 + except SQLAlchemyError: + return {"error": "Something went wrong while updating the database.", + "url": base_url}, 500 diff --git a/backend/tests/endpoints/courses_test.py b/backend/tests/endpoints/courses_test.py index 4df98cd5..ac299f01 100644 --- a/backend/tests/endpoints/courses_test.py +++ b/backend/tests/endpoints/courses_test.py @@ -25,19 +25,6 @@ def test_post_courses(self, courses_init_db, client, course_data, invalid_course assert course is not None assert course.teacher == "Bart" - response = client.post( - "/courses?uid=Jef", json=course_data - ) # non existent user - assert response.status_code == 404 - - response = client.post( - "/courses?uid=student_sel2_0", json=course_data - ) # existent user but no rights - assert response.status_code == 403 - - response = client.post("/courses", json=course_data) # bad link, no uid passed - assert response.status_code == 400 - response = client.post( "/courses?uid=Bart", json=invalid_course ) # invalid course @@ -88,11 +75,6 @@ def test_post_courses_course_id_students_and_admins(self, db_with_course, client sel2_admins_link = "/courses/" + str(course.course_id) + "/admins" - response = client.post( - sel2_admins_link + "?uid=student_sel2_0", # unauthorized user - json={"admin_uid": "Rien"}, - ) - assert response.status_code == 403 course_admins = [ s.uid for s in CourseAdmin.query.filter_by(course_id=course.course_id).all() @@ -131,7 +113,7 @@ def test_get_courses(self, courses_get_db, client, api_url): assert response.status_code == 200 sel2_students = [ - f"{api_url}/users/" + s.uid + {"uid": f"{api_url}/users/" + s.uid} for s in CourseStudent.query.filter_by(course_id=course.course_id).all() ]