Source code for inginious.frontend.plugin_manager

# -*- 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.

""" Plugin Manager """
import importlib
import logging
import bisect

from inginious.frontend.task_problems import get_displayable_problem_types
from inginious.common.tasks_problems import get_problem_types

[docs]class PluginManagerNotLoadedException(Exception): pass
[docs]class PluginManager(object): """ Registers an manage plugins. The init method inits only the Hook Manager; you have to call the method load() to start the plugins """ def __init__(self): self._logger = logging.getLogger("inginious.frontend.plugin_manager") self._hooks = {} self._loaded = False self._webpy_app = None self._flask_app = None self._task_factory = None self._database = None self._user_manager = None self._submission_manager = None def _exception_free_callback(self, callback, *args, **kwargs): """ A wrapper that remove all exceptions raised from hooks """ try: return callback(*args, **kwargs) except Exception: self._logger.exception("An exception occurred while calling a hook! ",exc_info=True) return None
[docs] def add_hook(self, name, callback, prio=0): """ Add a new hook that can be called with the call_hook function. `prio` is the priority. Higher priority hooks are called before lower priority ones. This function does not enforce a particular order between hooks with the same priorities. """ hook_list = self._hooks.get(name, []) add = (lambda *args, **kwargs: self._exception_free_callback(callback, *args, **kwargs)), -prio pos = bisect.bisect_right(list(x[1] for x in hook_list), -prio) hook_list[pos:pos] = [add] self._hooks[name] = hook_list
[docs] def call_hook(self, name, **kwargs): """ Call all hooks registered with this name. Returns a list of the returns values of the hooks (in the order the hooks were added)""" return [y for y in [x(**kwargs) for x, _ in self._hooks.get(name, [])] if y is not None]
[docs] def call_hook_recursive(self, name, **kwargs): """ Call all hooks registered with this name. Each hook receives as arguments the return value of the previous hook call, or the initial params for the first hook. As such, each hook must return a dictionary with the received (eventually modified) args. Returns the modified args.""" for x, _ in self._hooks.get(name, []): out = x(**kwargs) if out is not None: #ignore already reported failure kwargs = out return kwargs
[docs] def load(self, client, flask_app, course_factory, task_factory, database, user_manager, submission_manager, config): """ Loads the plugin manager. Must be done after the initialisation of the client """ self._flask_app = flask_app self._task_factory = task_factory self._database = database self._user_manager = user_manager self._submission_manager = submission_manager self._loaded = True for entry in config: module_name = entry["plugin_module"] module = importlib.import_module(module_name) """ Load Problem sub-classes """ pbl_types = get_problem_types(module_name) self._task_factory.set_problem_types(pbl_types) """ Load DisplayableProblem sub-classes """ displayable_pbl_types = get_displayable_problem_types(module_name) self._task_factory.set_problem_types(displayable_pbl_types) """ Initialize the module """ module.init(self, course_factory, client, entry)
[docs] def add_page(self, pattern, classname_or_viewfunc): """ Add a new page to the web application. Only available after that the Plugin Manager is loaded """ if not self._loaded: raise PluginManagerNotLoadedException() self._flask_app.add_url_rule("/<cookieless:sessionid>" + pattern[1:], view_func=classname_or_viewfunc)
[docs] def add_task_file_manager(self, task_file_manager): """ Add a task file manager. Only available after that the Plugin Manager is loaded """ if not self._loaded: raise PluginManagerNotLoadedException() self._task_factory.add_custom_task_file_manager(task_file_manager)
[docs] def register_auth_method(self, auth_method): """ Register a new authentication method name the name of the authentication method, typically displayed by the webapp input_to_display Only available after that the Plugin Manager is loaded """ if not self._loaded: raise PluginManagerNotLoadedException() self._user_manager.register_auth_method(auth_method)
[docs] def get_database(self): """ Returns the frontend database""" return self._database
[docs] def get_submission_manager(self): """ Returns the submission manager""" return self._submission_manager
[docs] def get_user_manager(self): """ Returns the user manager""" return self._user_manager