Manage courses (#307)

* feat: Manage courses

- Added tables of courses and usercourses
- Added some columns to exercise and a constraint
- Fixed the view template and user template
- Fixed the upload method
- Added tests
- Added migration
- Added to the navbar the Course dropdown choice

Author: orronai

lms/lmsdb/bootstrap.py

@@ -226,6 +226,27 @@ def _add_api_keys_to_users_table(table: Model, _column: Field) -> None:
             user.save()
 
 
+def _add_course_and_numbers_to_exercises_table(
+    table: Model, course: models.Course,
+) -> None:
+    log.info(
+        'Adding Course, Numbers for exercises, might take some extra time...',
+    )
+    with db_config.database.transaction():
+        for exercise in table:
+            exercise.number = exercise.id
+            exercise.course = course
+            exercise.save()
+
+
+def _create_usercourses_objects(table: Model, course: models.Course) -> None:
+    log.info('Adding UserCourse for all users, might take some extra time...')
+    UserCourse = models.UserCourse
+    with db_config.database.transaction():
+        for user in table:
+            UserCourse.create(user=user, course=course)
+
+
 def _add_uuid_to_users_table(table: Model, _column: Field) -> None:
     log.info('Adding UUIDs for all users, might take some extra time...')
     with db_config.database.transaction():
@@ -240,6 +261,32 @@ def _api_keys_migration() -> bool:
     return True
 
 
+def _last_course_viewed_migration() -> bool:
+    User = models.User
+    _add_not_null_column(User, User.last_course_viewed)
+    return True
+
+
+def _exercise_course_migration(course: models.Course) -> bool:
+    Exercise = models.Exercise
+    _create_usercourses_objects(models.User, course)
+    _add_course_and_numbers_to_exercises_table(Exercise, course)
+    return True
+
+
+def _add_exercise_course_id_and_number_columns_constraint() -> bool:
+    Exercise = models.Exercise
+    migrator = db_config.get_migrator_instance()
+    with db_config.database.transaction():
+        course_not_exists = _add_not_null_column(Exercise, Exercise.course)
+        number_not_exists = _add_not_null_column(Exercise, Exercise.number)
+        if course_not_exists and number_not_exists:
+            migrate(
+                migrator.add_index('exercise', ('course_id', 'number'), True),
+            )
+        db_config.database.commit()
+
+
 def _last_status_view_migration() -> bool:
     Solution = models.Solution
     _migrate_column_in_table_if_needed(Solution, Solution.last_status_view)
@@ -254,11 +301,15 @@ def _uuid_migration() -> bool:
 
 def main():
     with models.database.connection_context():
+        if models.database.table_exists(models.Exercise.__name__.lower()):
+            _add_exercise_course_id_and_number_columns_constraint()
+
         if models.database.table_exists(models.Solution.__name__.lower()):
             _last_status_view_migration()
 
         if models.database.table_exists(models.User.__name__.lower()):
             _api_keys_migration()
+            _last_course_viewed_migration()
             _uuid_migration()
 
         models.database.create_tables(models.ALL_MODELS, safe=True)
@@ -267,6 +318,9 @@ def main():
             models.create_basic_roles()
         if models.User.select().count() == 0:
             models.create_demo_users()
+        if models.Course.select().count() == 0:
+            course = models.create_basic_course()
+            _exercise_course_migration(course)
 
     text_fixer.fix_texts()
     import_tests.load_tests_from_path('/app_dir/notebooks-tests')

lms/lmsdb/models.py

@@ -5,8 +5,8 @@
 from collections import Counter
 from datetime import datetime
 from typing import (
-    Any, Dict, Iterable, List, Optional, TYPE_CHECKING, Tuple,
-    Type, Union, cast,
+    Any, Dict, Iterable, List, Optional,
+    TYPE_CHECKING, Tuple, Type, Union, cast,
 )
 from uuid import uuid4
 
@@ -132,21 +132,50 @@ def is_viewer(self) -> bool:
         return self.name == RoleOptions.VIEWER.value or self.is_manager
 
 
+class Course(BaseModel):
+    name = CharField(unique=True)
+    date = DateTimeField(default=datetime.now)
+    end_date = DateTimeField(null=True)
+    close_registration_date = DateTimeField(default=datetime.now)
+    invite_code = CharField(null=True)
+    is_public = BooleanField(default=False)
+
+    def has_user(self, user_id: int) -> bool:
+        return UserCourse.is_user_registered(user_id, self)
+
+    @classmethod
+    def fetch(cls, user: 'User') -> Iterable['Course']:
+        return (
+            cls
+            .select()
+            .join(UserCourse)
+            .where(UserCourse.user == user.id)
+            .order_by(Course.name.desc())
+        )
+
+    def __str__(self):
+        return f'{self.name}: {self.date} - {self.end_date}'
+
+
 class User(UserMixin, BaseModel):
     username = CharField(unique=True)
     fullname = CharField()
     mail_address = CharField(unique=True)
     password = CharField()
     role = ForeignKeyField(Role, backref='users')
     api_key = CharField()
+    last_course_viewed = ForeignKeyField(Course, null=True)
     uuid = UUIDField(default=uuid4, unique=True)
 
     def get_id(self):
         return str(self.uuid)
 
-    def is_password_valid(self, password):
+    def is_password_valid(self, password) -> bool:
         return check_password_hash(self.password, password)
 
+    def has_course(self, course_id: int) -> bool:
+        return UserCourse.is_user_registered(self, course_id)
+
     @classmethod
     def get_system_user(cls) -> 'User':
         instance, _ = cls.get_or_create(**{
@@ -168,6 +197,9 @@ def random_password(cls, stronger: bool = False) -> str:
     def get_notifications(self) -> Iterable['Notification']:
         return Notification.fetch(self)
 
+    def get_courses(self) -> Iterable['Course']:
+        return Course.fetch(self)
+
     def notes(self) -> Iterable['Note']:
         fields = (
             Note.id, Note.creator.fullname, CommentText.text,
@@ -215,6 +247,24 @@ def on_save_handler(model_class, instance, created):
         instance.api_key = generate_password_hash(instance.api_key)
 
 
+class UserCourse(BaseModel):
+    user = ForeignKeyField(User, backref='usercourses')
+    course = ForeignKeyField(Course, backref='usercourses')
+    date = DateTimeField(default=datetime.now)
+
+    @classmethod
+    def is_user_registered(cls, user_id: int, course_id: int) -> bool:
+        return (
+            cls.
+            select()
+            .where(
+                cls.user == user_id,
+                cls.course == course_id,
+            )
+            .exists()
+        )
+
+
 class Notification(BaseModel):
     ID_FIELD_NAME = 'id'
     MAX_PER_USER = 10
@@ -304,15 +354,46 @@ class Exercise(BaseModel):
     due_date = DateTimeField(null=True)
     notebook_num = IntegerField(default=0)
     order = IntegerField(default=0, index=True)
+    course = ForeignKeyField(Course, backref='exercise')
+    number = IntegerField(default=1)
+
+    class Meta:
+        indexes = (
+            (('course_id', 'number'), True),
+        )
 
     def open_for_new_solutions(self) -> bool:
         if self.due_date is None:
             return not self.is_archived
         return datetime.now() < self.due_date and not self.is_archived
 
     @classmethod
-    def get_objects(cls, fetch_archived: bool = False):
-        exercises = cls.select().order_by(Exercise.order)
+    def get_highest_number(cls):
+        return cls.select(fn.MAX(cls.number)).scalar()
+
+    @classmethod
+    def is_number_exists(cls, number: int) -> bool:
+        return cls.select().where(cls.number == number).exists()
+
+    @classmethod
+    def get_objects(
+        cls, user_id: int, fetch_archived: bool = False,
+        from_all_courses: bool = False,
+    ):
+        user = User.get(User.id == user_id)
+        exercises = (
+            cls
+            .select()
+            .join(Course)
+            .join(UserCourse)
+            .where(UserCourse.user == user_id)
+            .switch()
+            .order_by(UserCourse.date, Exercise.number, Exercise.order)
+        )
+        if not from_all_courses:
+            exercises = exercises.where(
+                UserCourse.course == user.last_course_viewed,
+            )
         if not fetch_archived:
             exercises = exercises.where(cls.is_archived == False)  # NOQA: E712
         return exercises
@@ -324,6 +405,9 @@ def as_dict(self) -> Dict[str, Any]:
             'is_archived': self.is_archived,
             'notebook': self.notebook_num,
             'due_date': self.due_date,
+            'exercise_number': self.number,
+            'course_id': self.course.id,
+            'course_name': self.course.name,
         }
 
     @staticmethod
@@ -334,6 +418,14 @@ def __str__(self):
         return self.subject
 
 
+@pre_save(sender=Exercise)
+def exercise_number_save_handler(model_class, instance, created):
+    """Change the exercise number to the highest consecutive number."""
+
+    if model_class.is_number_exists(instance.number):
+        instance.number = model_class.get_highest_number() + 1
+
+
 class SolutionState(enum.Enum):
     CREATED = 'Created'
     IN_CHECKING = 'In checking'
@@ -469,10 +561,13 @@ def test_results(self) -> Iterable[dict]:
     @classmethod
     def of_user(
         cls, user_id: int, with_archived: bool = False,
+        from_all_courses: bool = False,
     ) -> Iterable[Dict[str, Any]]:
-        db_exercises = Exercise.get_objects(fetch_archived=with_archived)
+        db_exercises = Exercise.get_objects(
+            user_id=user_id, fetch_archived=with_archived,
+            from_all_courses=from_all_courses,
+        )
         exercises = Exercise.as_dicts(db_exercises)
-
         solutions = (
             cls
             .select(cls.exercise, cls.id, cls.state, cls.checker)
@@ -946,7 +1041,7 @@ def generate_string(
     return ''.join(password)
 
 
-def create_demo_users():
+def create_demo_users() -> None:
     print('First run! Here are some users to get start with:')  # noqa: T001
     fields = ['username', 'fullname', 'mail_address', 'role']
     student_role = Role.by_name('Student')
@@ -964,9 +1059,13 @@ def create_demo_users():
         print(f"User: {user['username']}, Password: {password}")  # noqa: T001
 
 
-def create_basic_roles():
+def create_basic_roles() -> None:
     for role in RoleOptions:
         Role.create(name=role.value)
 
 
+def create_basic_course() -> Course:
+    return Course.create(name='Python Course', date=datetime.now())
+
+
 ALL_MODELS = BaseModel.__subclasses__()