feat: courses refactor

src/services/courses/courses.py

@@ -0,0 +1,212 @@
+import json
+import os
+from typing import List
+from uuid import uuid4
+from pydantic import BaseModel
+from src.services.uploads import upload_thumbnail
+from src.services.users import PublicUser, User
+from src.services.database import create_config_collection, check_database, create_database, learnhouseDB, learnhouseDB
+from src.services.security import *
+from fastapi import FastAPI, HTTPException, status, Request, Response, BackgroundTasks, UploadFile, File
+from datetime import datetime
+
+#### Classes ####################################################
+
+
+class Course(BaseModel):
+    name: str
+    mini_description: str
+    description: str
+    learnings: List[str]
+    thumbnail: str
+    public: bool
+    chapters: List[str]
+    org_id: str
+
+
+class CourseInDB(Course):
+    course_id: str
+    creationDate: str
+    updateDate: str
+    authors: List[str]
+
+#####
+
+
+
+
+#### Classes ####################################################
+
+# TODO : Add courses photo & cover upload and delete
+
+
+# Courses
+
+async def get_course(course_id: str, current_user: PublicUser):
+    await check_database()
+    courses = learnhouseDB["courses"]
+
+    course = courses.find_one({"course_id": course_id})
+
+    # verify course rights
+    await verify_rights(course_id, current_user, "read")
+
+    if not course:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
+
+    course = Course(**course)
+    return course
+
+
+async def create_course(course_object: Course,   org_id: str, current_user: PublicUser, thumbnail_file: UploadFile | None = None):
+    await check_database()
+    courses = learnhouseDB["courses"]
+
+    # generate course_id with uuid4
+    course_id = str(f"course_{uuid4()}")
+
+    # TODO(fix) : the implementation here is clearly not the best one (this entire function)
+    course_object.org_id = org_id
+    hasRoleRights = await verify_user_rights_with_roles("create", current_user.user_id, course_id)
+
+    if not hasRoleRights:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Roles : Insufficient rights to perform this action")
+
+    if thumbnail_file:
+        name_in_disk = f"{course_id}_thumbnail_{uuid4()}.{thumbnail_file.filename.split('.')[-1]}"
+        await upload_thumbnail(thumbnail_file, name_in_disk)
+        course_object.thumbnail = name_in_disk
+
+    course = CourseInDB(course_id=course_id, authors=[
+        current_user.user_id], creationDate=str(datetime.now()), updateDate=str(datetime.now()), **course_object.dict())
+
+    course_in_db = courses.insert_one(course.dict())
+
+    if not course_in_db:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Unavailable database")
+
+    return course.dict()
+
+
+async def update_course_thumbnail(course_id: str, current_user: PublicUser, thumbnail_file: UploadFile | None = None):
+    await check_database()
+
+    # verify course rights
+    await verify_rights(course_id, current_user, "update")
+
+    courses = learnhouseDB["courses"]
+
+    course = courses.find_one({"course_id": course_id})
+    # TODO(fix) : the implementation here is clearly not the best one
+    if course:
+        creationDate = course["creationDate"]
+        authors = course["authors"]
+        if thumbnail_file:
+            name_in_disk = f"{course_id}_thumbnail_{uuid4()}.{thumbnail_file.filename.split('.')[-1]}"
+            course = Course(**course).copy(update={"thumbnail": name_in_disk})
+            await upload_thumbnail(thumbnail_file, name_in_disk)
+
+            updated_course = CourseInDB(course_id=course_id, creationDate=creationDate,
+                                        authors=authors, updateDate=str(datetime.now()), **course.dict())
+
+            courses.update_one({"course_id": course_id}, {
+                "$set": updated_course.dict()})
+
+            return CourseInDB(**updated_course.dict())
+
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
+
+
+async def update_course(course_object: Course, course_id: str, current_user: PublicUser):
+    await check_database()
+
+    # verify course rights
+    await verify_rights(course_id, current_user, "update")
+
+    courses = learnhouseDB["courses"]
+
+    course = courses.find_one({"course_id": course_id})
+
+    if course:
+        creationDate = course["creationDate"]
+        authors = course["authors"]
+
+        # get today's date
+        datetime_object = datetime.now()
+
+        updated_course = CourseInDB(
+            course_id=course_id, creationDate=creationDate, authors=authors, updateDate=str(datetime_object), **course_object.dict())
+
+        courses.update_one({"course_id": course_id}, {
+            "$set": updated_course.dict()})
+
+        return CourseInDB(**updated_course.dict())
+
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
+
+
+async def delete_course(course_id: str, current_user: PublicUser):
+    await check_database()
+
+    # verify course rights
+    await verify_rights(course_id, current_user, "delete")
+
+    courses = learnhouseDB["courses"]
+
+    course = courses.find_one({"course_id": course_id})
+
+    if not course:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail="Course does not exist")
+
+    isDeleted = courses.delete_one({"course_id": course_id})
+
+    if isDeleted:
+        return {"detail": "Course deleted"}
+    else:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Unavailable database")
+
+
+async def get_courses(page: int = 1, limit: int = 10, org_id: str | None = None):
+    await check_database()
+    courses = learnhouseDB["courses"]
+    # TODO : Get only courses that user is admin/has roles of
+    # get all courses from database
+    all_courses = courses.find({"org_id": org_id}).sort(
+        "name", 1).skip(10 * (page - 1)).limit(limit)
+
+    return [json.loads(json.dumps(course, default=str)) for course in all_courses]
+
+
+
+#### Security ####################################################
+
+
+async def verify_rights(course_id: str, current_user: PublicUser, action: str):
+    await check_database()
+    courses = learnhouseDB["courses"]
+
+    course = courses.find_one({"course_id": course_id})
+
+    if not course:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT, detail=f"Course/CourseChapter does not exist")
+
+    hasRoleRights = await verify_user_rights_with_roles(action, current_user.user_id, course_id)
+    isAuthor = current_user.user_id in course["authors"]
+
+    if not hasRoleRights and not isAuthor:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail="Roles/Ownership : Insufficient rights to perform this action")
+
+    return True
+
+#### Security ####################################################