flaschengeist/flaschengeist/controller/pluginController.py

"""Controller for Plugin logic

Used by plugins for setting and notification functionality.
"""

import sqlalchemy

from typing import Union
from flask import current_app
from werkzeug.exceptions import NotFound
from sqlalchemy.exc import OperationalError
from importlib.metadata import entry_points

from .. import logger
from ..database import db
from ..utils.hook import Hook
from ..models import Plugin, PluginSetting, Notification


def get_enabled_plugins():
    try:
        enabled_plugins = Plugin.query.filter(Plugin.enabled == True).all()
    except OperationalError as e:

        class PluginStub:
            def __init__(self, name) -> None:
                self.name = name
                self.version = "?"

        logger.error("Could not connect to database or database not initialized! No plugins enabled!")
        logger.debug("Can not query enabled plugins", exc_info=True)
        enabled_plugins = [
            PluginStub("auth"),
            PluginStub("roles"),
            PluginStub("users"),
            PluginStub("scheduler"),
        ]
    return enabled_plugins


def get_setting(plugin_id: str, name: str, **kwargs):
    """Get plugin setting from database

    Args:
        plugin_id: ID of the plugin
        name: string identifying the setting
        default: Default value
    Returns:
        Value stored in database (native python)
    Raises:
        `KeyError` if no such setting exists in the database
    """
    try:
        setting = PluginSetting.query.filter(PluginSetting.plugin == plugin_id).filter(PluginSetting.name == name).one()
        return setting.value
    except sqlalchemy.orm.exc.NoResultFound:
        if "default" in kwargs:
            return kwargs["default"]
        else:
            raise KeyError


def set_setting(plugin_id: str, name: str, value):
    """Save setting in database

    Args:
        plugin_id: ID of the plugin
        name: String identifying the setting
        value: Value to be stored
    """
    setting = (
        PluginSetting.query.filter(PluginSetting.plugin == plugin_id).filter(PluginSetting.name == name).one_or_none()
    )
    if setting is not None:
        if value is None:
            db.session.delete(setting)
        else:
            setting.value = value
    else:
        db.session.add(PluginSetting(plugin=plugin_id, name=name, value=value))
    db.session.commit()


def notify(plugin_id: str, user, text: str, data=None):
    """Create a new notification for an user

    Args:
        plugin_id: ID of the plugin
        user: `flaschengeist.models.user.User` to notify
        text: Visibile notification text
        data: Optional data passed to the notificaton
    Returns:
        ID of the created `flaschengeist.models.notification.Notification`

    Hint: use the data for frontend actions.
    """
    if not user.deleted:
        n = Notification(text=text, data=data, plugin=plugin_id, user_=user)
        db.session.add(n)
        db.session.commit()
        return n.id


@Hook("plugins.installed")
def install_plugin(plugin_name: str):
    logger.debug(f"Installing plugin {plugin_name}")
    entry_point = entry_points(group="flaschengeist.plugins", name=plugin_name)
    if not entry_point:
        raise NotFound

    plugin = entry_point[0].load()(entry_point[0])
    entity = Plugin(name=plugin.name, version=plugin.version)
    db.session.add(entity)
    db.session.commit()
    return entity


@Hook("plugin.uninstalled")
def uninstall_plugin(plugin_id: Union[str, int]):
    plugin = disable_plugin(plugin_id)
    logger.debug(f"Uninstall plugin {plugin.name}")

    entity = current_app.config["FG_PLUGINS"][plugin.name]
    entity.uninstall()
    del current_app.config["FG_PLUGINS"][plugin.name]
    db.session.delete(plugin)
    db.session.commit()


@Hook("plugins.enabled")
def enable_plugin(plugin_id: Union[str, int]):
    logger.debug(f"Enabling plugin {plugin_id}")
    plugin: Plugin = Plugin.query
    if isinstance(plugin_id, str):
        plugin = plugin.filter(Plugin.name == plugin_id).one_or_none()
        if plugin is None:
            logger.debug("Plugin not installed, trying to install")
            plugin = install_plugin(plugin_id)
    else:
        plugin = plugin.get(plugin_id)
        if plugin is None:
            raise NotFound
    plugin.enabled = True
    db.session.commit()

    return plugin


@Hook("plugins.disabled")
def disable_plugin(plugin_id: Union[str, int]):
    logger.debug(f"Disabling plugin {plugin_id}")
    plugin: Plugin = Plugin.query
    if isinstance(plugin_id, str):
        plugin = plugin.filter(Plugin.name == plugin_id).one_or_none()
    else:
        plugin = plugin.get(plugin_id)
    if plugin is None:
        raise NotFound
    plugin.enabled = False
    db.session.commit()

    return plugin