Source code for inginious.common.course_factory

# -*- coding: utf-8 -*-
#
# This file is part of INGInious. See the LICENSE and the COPYRIGHTS files for
# more information about the licensing of this file.

""" Factory for loading courses from disk """
from inginious.common.filesystems.provider import FileSystemProvider
from inginious.common.log import get_course_logger
from inginious.common.courses import Course
from inginious.common.base import id_checker, get_json_or_yaml, loads_json_or_yaml
from inginious.common.task_factory import TaskFactory
from inginious.common.tasks import Task
from inginious.common.hook_manager import HookManager
from inginious.common.exceptions import InvalidNameException, CourseNotFoundException, CourseUnreadableException, CourseAlreadyExistsException


[docs]class CourseFactory(object): """ Load courses from disk """ def __init__(self, filesystem: FileSystemProvider, task_factory, hook_manager, course_class=Course): self._filesystem = filesystem self._task_factory = task_factory self._hook_manager = hook_manager self._course_class = course_class self._cache = {}
[docs] def get_course(self, courseid): """ :param courseid: the course id of the course :raise InvalidNameException, CourseNotFoundException, CourseUnreadableException :return: an object representing the course, of the type given in the constructor """ if not id_checker(courseid): raise InvalidNameException("Course with invalid name: " + courseid) if self._cache_update_needed(courseid): self._update_cache(courseid) return self._cache[courseid][0]
[docs] def get_task(self, courseid, taskid): """ Shorthand for CourseFactory.get_course(courseid).get_task(taskid) :param courseid: the course id of the course :param taskid: the task id of the task :raise InvalidNameException, CourseNotFoundException, CourseUnreadableException, TaskNotFoundException, TaskUnreadableException :return: an object representing the task, of the type given in the constructor """ return self.get_course(courseid).get_task(taskid)
[docs] def get_task_factory(self): """ :return: the associated task factory """ return self._task_factory
[docs] def get_course_descriptor_content(self, courseid): """ :param courseid: the course id of the course :raise InvalidNameException, CourseNotFoundException, CourseUnreadableException :return: the content of the dict that describes the course """ path = self._get_course_descriptor_path(courseid) return loads_json_or_yaml(path, self._filesystem.get(path).decode("utf-8"))
[docs] def update_course_descriptor_content(self, courseid, content): """ Updates the content of the dict that describes the course :param courseid: the course id of the course :param content: the new dict that replaces the old content :raise InvalidNameException, CourseNotFoundException """ path = self._get_course_descriptor_path(courseid) self._filesystem.put(path, get_json_or_yaml(path, content))
[docs] def get_course_fs(self, courseid): """ :param courseid: :return: a FileSystemProvider pointing to the directory of the course """ if not id_checker(courseid): raise InvalidNameException("Course with invalid name: " + courseid) return self._filesystem.from_subfolder(courseid)
[docs] def get_all_courses(self): """ :return: a table containing courseid=>Course pairs """ course_ids = [f[0:len(f)-1] for f in self._filesystem.list(folders=True, files=False, recursive=False)] # remove trailing "/" output = {} for courseid in course_ids: try: output[courseid] = self.get_course(courseid) except Exception: get_course_logger(courseid).warning("Cannot open course", exc_info=True) return output
def _get_course_descriptor_path(self, courseid): """ :param courseid: the course id of the course :raise InvalidNameException, CourseNotFoundException :return: the path to the descriptor of the course """ if not id_checker(courseid): raise InvalidNameException("Course with invalid name: " + courseid) course_fs = self.get_course_fs(courseid) if course_fs.exists("course.yaml"): return courseid+"/course.yaml" if course_fs.exists("course.json"): return courseid+"/course.json" raise CourseNotFoundException()
[docs] def create_course(self, courseid, init_content): """ :param courseid: the course id of the course :param init_content: initial descriptor content :raise InvalidNameException or CourseAlreadyExistsException Create a new course folder and set initial descriptor content, folder can already exist """ if not id_checker(courseid): raise InvalidNameException("Course with invalid name: " + courseid) course_fs = self.get_course_fs(courseid) course_fs.ensure_exists() if course_fs.exists("course.yaml") or course_fs.exists("course.json"): raise CourseAlreadyExistsException("Course with id " + courseid + " already exists.") else: course_fs.put("course.yaml", get_json_or_yaml("course.yaml", init_content)) get_course_logger(courseid).info("Course %s created in the factory.", courseid)
[docs] def delete_course(self, courseid): """ :param courseid: the course id of the course :raise InvalidNameException or CourseNotFoundException Erase the content of the course folder """ if not id_checker(courseid): raise InvalidNameException("Course with invalid name: " + courseid) course_fs = self.get_course_fs(courseid) if not course_fs.exists(): raise CourseNotFoundException() course_fs.delete() get_course_logger(courseid).info("Course %s erased from the factory.", courseid)
def _cache_update_needed(self, courseid): """ :param courseid: the (valid) course id of the course :raise InvalidNameException, CourseNotFoundException :return: True if an update of the cache is needed, False else """ if courseid not in self._cache: return True try: descriptor_name = self._get_course_descriptor_path(courseid) last_update = {descriptor_name: self._filesystem.get_last_modification_time(descriptor_name)} translations_fs = self._filesystem.from_subfolder("$i18n") if translations_fs.exists(): for f in translations_fs.list(folders=False, files=True, recursive=False): lang = f[0:len(f) - 3] if translations_fs.exists(lang + ".mo"): last_update["$i18n/" + lang + ".mo"] = translations_fs.get_last_modification_time(lang + ".mo") except: raise CourseNotFoundException() last_modif = self._cache[courseid][1] for filename, mftime in last_update.items(): if filename not in last_modif or last_modif[filename] < mftime: return True return False def _update_cache(self, courseid): """ Updates the cache :param courseid: the (valid) course id of the course :raise InvalidNameException, CourseNotFoundException, CourseUnreadableException """ path_to_descriptor = self._get_course_descriptor_path(courseid) try: course_descriptor = loads_json_or_yaml(path_to_descriptor, self._filesystem.get(path_to_descriptor).decode("utf8")) except Exception as e: raise CourseUnreadableException(str(e)) last_modif = {path_to_descriptor: self._filesystem.get_last_modification_time(path_to_descriptor)} translations_fs = self._filesystem.from_subfolder("$i18n") if translations_fs.exists(): for f in translations_fs.list(folders=False, files=True, recursive=False): lang = f[0:len(f) - 3] if translations_fs.exists(lang + ".mo"): last_modif["$i18n/" + lang + ".mo"] = translations_fs.get_last_modification_time(lang + ".mo") self._cache[courseid] = ( self._course_class(courseid, course_descriptor, self.get_course_fs(courseid), self._task_factory, self._hook_manager), last_modif ) self._task_factory.update_cache_for_course(courseid)
[docs]def create_factories(fs_provider, task_problem_types, hook_manager=None, course_class=Course, task_class=Task): """ Shorthand for creating Factories :param fs_provider: A FileSystemProvider leading to the courses :param hook_manager: an Hook Manager instance. If None, a new Hook Manager is created :param course_class: :param task_class: :return: a tuple with two objects: the first being of type CourseFactory, the second of type TaskFactory """ if hook_manager is None: hook_manager = HookManager() task_factory = TaskFactory(fs_provider, hook_manager, task_problem_types, task_class) return CourseFactory(fs_provider, task_factory, hook_manager, course_class), task_factory