Implement course APIs

backend/blueprints/course/__init__.py

@@ -1,18 +1,112 @@
-from flask import Blueprint, request, jsonify, g
+from flask import Blueprint, request, jsonify, g, url_for
-from sqlalchemy import select, and_
+from sqlalchemy import select, and_, func, distinct, or_
+from sqlalchemy.exc import IntegrityError
 from werkzeug.datastructures import MultiDict
 import os
 import uuid
+import math
 from config import DEFAULT_COURSE_COVER
-from db.model import db, Course, Category, User, Chat
+from db.model import db, Course, Category, User, Chat, Enrollment
 from utils.utils import random_string_generator
 from utils.auth import auth_required, requires_role
 from constants import *
 from config import *
 from constants import PublishedStatus
+from typing import Union
+from backend.constants import UserRole
 
 course = Blueprint('course', __name__)
 
+@course.route('/listAll')
+def list_all_courses():
+    limit: int = int(request.args.get('limit', 10))
+    offset: int = int(request.args.get('offset', 1))
+    category_uuid: str = request.args.get('category_uuid')
+    search_q: str = request.args.get('search_q', '').strip()
+    sort_by: str = request.args.get('sort_by', '').strip()
+    available_sorts = ['date_asc', 'date_desc', 'name_asc', 'name_desc', 'students_desc', 'students_asc']
+    if category_uuid is not None:
+        category_uuid: uuid.UUID = uuid.UUID(request.args.get('category_uuid'))
+    # Build the query as required
+    query: select = select(Course).limit(limit).offset(offset)
+    if search_q != '':
+        query = query.where(or_(Course.name.like(f'%{search_q}%'), Course.description.like(f'%{search_q}%'),
+                                Course.author.firstName.like(f'%{search_q}%')))
+    if category_uuid is not None:
+        query = query.where(Course.categoryID == category_uuid)
+    total_pages_for_offset: int = db.session.execute(func.count(Course.id).select_from(Course)).scalar()/limit
+    total_pages: int = math.ceil(total_pages_for_offset)
+    if sort_by in available_sorts:
+        if sort_by == 'date_asc':
+            query = query.order_by(Course.creationDate.asc())
+        elif sort_by == 'date_desc':
+            query = query.order_by(Course.creationDate.desc())
+        elif sort_by == 'name_asc':
+            query = query.order_by(Course.name.asc())
+        elif sort_by == 'name_desc':
+            query = query.order_by(Course.name.desc())
+        elif sort_by == 'students_desc':
+            query = query.order_by(Course.totalEnrolled.desc())
+        elif sort_by == 'students_asc':
+            query = query.order_by(Course.totalEnrolled.asc())
+    courses: list[Course] = db.session.execute(query).scalars()
+    course_list: list[dict] = []
+    for item in courses:
+        course_list.append(
+            {
+                'id': item.id,
+                'name': item.name,
+                'description': item.description,
+                'isActive': item.isActive,
+                'creationDate': item.creationDate,
+                'coverImage': url_for('send_file', filename=item.coverImage),
+                'totalEnrolled': item.totalEnrolled,
+                'author': {
+                    'id': item.author.id,
+                    'firstName': item.author.firstName,
+                    'lastName': item.author.LastName,
+                    'username': item.author.username,
+                    'bio': item.author.bio,
+                    'lastOnline': item.author.lastOnline,
+                    'pfpFilename': url_for('send_file', filename=item.author.pfpFilename)
+                },
+                'category': {
+                    'id': item.categoryID,
+                    'name': item.category.name,
+                    'description': item.category.description
+                }
+            })
+    return jsonify({
+        'total_pages': total_pages,
+        'current_offset': offset,
+        'limit': limit,
+        'data': course_list,
+    })
+
+@course.route('/enroll')
+@auth_required()
+def enroll_user():
+    if not request.form.get('course_uuid'):
+        return jsonify({'message': 'Missing required parameter "course_uuid" '}), 400
+    course_uuid: uuid.UUID = uuid.UUID(request.form.get('course_uuid'))
+    selected_course: Course = db.session.execute(select(Course).where(Course.id == course_uuid)).scalar()
+    if not selected_course:
+        return jsonify({'message': 'Course not found'}), 404
+    new_enroll: Enrollment = Enrollment(
+        userID=g.current_user.id,
+        courseID=course_uuid
+    )
+    try:
+        selected_course.totalEnrolled = selected_course.totalEnrolled + 1
+        db.session.add(new_enroll)
+        db.session.commit()
+    except IntegrityError:
+        return jsonify({'message': 'Already enrolled to this course'})
+    return jsonify({'message': 'Enrollment successful'}), 200
+
+
+
+
 @course.route('/create', methods=['POST'])
 @auth_required()
 def create_course():
@@ -32,7 +126,6 @@ def create_course():
         course_name: str = form_data['course_name']
     except KeyError:
         return jsonify({'message': 'Course name cannot be empty'}), 401
-
     course_description: str = form_data.get('course_description', '')
     category_id: uuid.UUID = uuid.UUID(form_data['category_uuid'])
     pages_required_for_community: int = int(form_data['community_unlock_at_pages'])  # TODO: Add this field to model
@@ -124,4 +217,132 @@ def update_course():
 
 @course.route('/info/<string:course_uuid>')
 def course_info(course_uuid):
+    course_uuid: uuid.UUID = uuid.UUID(course_uuid)
+    selected_course: Course = db.session.execute(select(Course).where(and_(Course.id == course_uuid))).scalar()
+    if not selected_course:
+        return jsonify({'message': 'The course does not exist'}), 404
+    # Only allow owner or admin to query info for course that is not published or is pending
+    if not selected_course.isActive or selected_course.publishedStatus != int(PublishedStatus.APPROVED):
+        if g.get("is_authed"):
+            if g.current_user.role == int(UserRole.ADMIN) or g.current_user.id == selected_course.authorID
                 pass
+            else:
+                return jsonify({'message': 'The course does not exist.'}), 404
+    self_enrollment_record: Union[None, Enrollment] = None
+    self_enrollment_data: dict = {}
+    if g.get("is_authed"):
+        self_enrollment_record: Enrollment = db.session.execute(
+            select(Enrollment).where(
+                and_(
+                    Enrollment.courseID == selected_course.id, Enrollment.userID == g.current_user.id
+                )
+            )
+        )
+        if self_enrollment_record:
+            self_enrollment_data = {
+                'lastActivity': self_enrollment_record.lastActivity,
+                'currentPage': self_enrollment_record.currentPage,
+                'maxPage': self_enrollment_record.maxPage,
+                'joinedDate': self_enrollment_record.joinedDate,
+                'userID': self_enrollment_record.userID
+            }
+        # Get total enrolled user and total unique user chatting about the course and put it in dict
+        summary_user: dict = {
+            'totalEnrolled': db.session.execute(
+                select(func.count(Enrollment.id)).where(Enrollment.courseID == course_uuid)
+            ).scalar(),
+
+            'usersInChat': db.session.execute(
+                select(func.count(distinct(Chat.userID))).select_from(Chat).where(Chat.courseID == course_uuid)
+            ).scalar(),
+
+            'totalChats': db.session.execute(
+                select(func.count()).select_from(Chat).where(Chat.courseID == course_uuid)
+            ).scalar()
+        }
+    jsonify({
+        'message': 'successful',
+        'data': {
+            'id':  selected_course.id,
+            'courseName': selected_course.name,
+            'courseDescription': selected_course.description,
+            'isActive':  selected_course.isActive,
+            'publishedStatus': selected_course.publishedStatus,
+            'creationDate': selected_course.creationDate, # TODO: Format to particular structure
+            'coverImage': url_for('send_file', filename=selected_course.coverImage),
+            'serverFilename': url_for('send_file', filename=selected_course.serverFilename),
+            'totalPages': 100,
+            'author': {
+                'id': selected_course.authorID,
+                'username': selected_course.author.username,
+                'firstName': selected_course.author.firstName,
+                'lastName': selected_course.author.lastName,
+                'pfpFilename': url_for('send_file', filename=selected_course.author.pfpFilename),
+                'bio':  selected_course.author.bio
+            },
+            'selfEnrollment': {
+                'isEnrolled': self_enrollment_record is not None,
+                'data': self_enrollment_data
+            },
+            'enrollmentSummary':  summary_user
+        }
+    }), 200
+
+@course.route('/getCategories', methods=['GET'])
+def get_categories():
+    categories: list[Category] = db.session.execute(select(Category)).scalars()
+    cat_list: list[dict] = []
+    for category in categories:
+        cat_list.append(
+            {
+                'id': category.id,
+                'name': category.name,
+                'description': category.description,
+                'isActive': category.isActive,
+                'creationDate': category.creationDate
+            }
+        )
+    return jsonify(cat_list), 200
+
+@course.route('/createCategory', methods=['Post'])
+@auth_required()
+@requires_role([UserRole.ADMIN])
+def create_category():
+    try:
+        new_cat: Category = Category(
+        name=request.form['name'],
+        description=request.form.get('description'),
+        isActive=bool(int(request.form.get('isActive'))),
+        courses=[]
+        )
+    except KeyError:
+        return jsonify({'message': 'Missing required parameter "name" '}), 400
+    db.session.add(new_cat)
+    db.session.commit()
+    return jsonify({'message': 'Category created'}), 201
+
+@course.route('/updateCategory', methods=['POST', 'DELETE'])
+@auth_required()
+@requires_role([UserRole.ADMIN])
+def update_category():
+    form_data: dict = request.form
+    try:
+        category_id: uuid.UUID = uuid.UUID(form_data['category_id'])
+    except KeyError:
+        return jsonify({'message': 'Missing required parameter "category_id" '}), 400
+    selected_category: Category = db.session.execute(select(Category).where(Category.id == category_id)).scalar()
+    if not selected_category:
+        return jsonify({'message': 'Category not found'}), 404
+    if request.method == 'DELETE':
+        db.session.delete(selected_category)
+        db.session.commit()
+        return jsonify({'message': 'Category deleted'}), 200
+    else:
+        if form_data.get('name'):
+            selected_category.name = form_data.get('name')
+        if form_data.get('description'):
+            selected_category.description = form_data.get('description')
+        if form_data.get('isActive'):
+            selected_category.isActive = bool(int(form_data.get('isActive')))
+        db.session.commit()
+        return jsonify({'message': 'Category updated'}), 200

backend/db/model.py

@@ -1,6 +1,6 @@
 from flask_sqlalchemy import SQLAlchemy
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, MappedAsDataclass
-from sqlalchemy import types, text, String, DateTime, func, Boolean, ForeignKey, SmallInteger
+from sqlalchemy import types, text, String, DateTime, func, Boolean, ForeignKey, SmallInteger, Integer
 from datetime import datetime
 import uuid
 from typing import List
@@ -73,6 +73,7 @@ class Course(db.Model):
     authorID: Mapped[uuid.UUID] = mapped_column(ForeignKey("user.id"))
     author: Mapped["User"] = relationship(back_populates="publications")
     description: Mapped[str] = mapped_column(String(1024), nullable=False, default='')
+    totalEnrolled: Mapped[int] = mapped_column(Integer, nullable=False, default='')
     isActive: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
     publishedStatus: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=PublishedStatus.DRAFT)
     creationDate: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=func.now())