flaschengeist/flaschengeist/app.py

import enum

import pkg_resources
from flask import Flask, current_app
from flask_cors import CORS
from datetime import datetime, date
from flask.json import JSONEncoder, jsonify
from sqlalchemy.exc import OperationalError
from werkzeug.exceptions import HTTPException

from . import logger
from .plugins import AuthPlugin
from flaschengeist.config import config, configure_app
from flaschengeist.controller import roleController
from flaschengeist.utils.hook import Hook


class CustomJSONEncoder(JSONEncoder):
    def default(self, o):
        try:
            # Check if custom model
            return o.serialize()
        except AttributeError:
            pass
        if isinstance(o, datetime) or isinstance(o, date):
            return o.isoformat()
        if isinstance(o, enum.Enum):
            return o.value
        try:
            # Check if iterable
            iterable = iter(o)
        except TypeError:
            pass
        else:
            return list(iterable)
        return JSONEncoder.default(self, o)


@Hook("plugins.loaded")
def __load_plugins(app):
    logger.debug("Search for plugins")

    app.config["FG_PLUGINS"] = {}
    for entry_point in pkg_resources.iter_entry_points("flaschengeist.plugins"):
        logger.debug(f"Found plugin: >{entry_point.name}<")

        if entry_point.name == config["FLASCHENGEIST"]["auth"] or (
            entry_point.name in config and config[entry_point.name].get("enabled", False)
        ):
            logger.debug(f"Load plugin {entry_point.name}")
            try:
                plugin = entry_point.load()
                if not hasattr(plugin, "name"):
                    setattr(plugin, "name", entry_point.name)
                plugin = plugin(config.get(entry_point.name, {}))
                if hasattr(plugin, "blueprint") and plugin.blueprint is not None:
                    app.register_blueprint(plugin.blueprint)
            except:
                logger.error(
                    f"Plugin {entry_point.name} was enabled, but could not be loaded due to an error.",
                    exc_info=True,
                )
                continue
            if isinstance(plugin, AuthPlugin):
                if entry_point.name != config["FLASCHENGEIST"]["auth"]:
                    logger.debug(f"Unload not configured AuthPlugin {entry_point.name}")
                    del plugin
                    continue
                else:
                    logger.info(f"Using authentication plugin: {entry_point.name}")
                    app.config["FG_AUTH_BACKEND"] = plugin
            else:
                logger.info(f"Using plugin: {entry_point.name}")
            app.config["FG_PLUGINS"][entry_point.name] = plugin
        else:
            logger.debug(f"Skip disabled plugin {entry_point.name}")
    if "FG_AUTH_BACKEND" not in app.config:
        logger.error("No authentication plugin configured or authentication plugin not found")
        raise RuntimeError("No authentication plugin configured or authentication plugin not found")


@Hook("plugins.installed")
def install_all():
    from flaschengeist.database import db

    db.create_all()
    db.session.commit()
    for name, plugin in current_app.config["FG_PLUGINS"].items():
        if not plugin:
            logger.debug(f"Skip disabled plugin: {name}")
            continue
        logger.info(f"Install plugin {name}")
        plugin.install()
        if plugin.permissions:
            roleController.create_permissions(plugin.permissions)


def create_app(test_config=None, cli=False):
    app = Flask(__name__)
    app.json_encoder = CustomJSONEncoder
    CORS(app)

    with app.app_context():
        from flaschengeist.database import db

        configure_app(app, test_config, cli)
        db.init_app(app)
        __load_plugins(app)

    @app.route("/", methods=["GET"])
    def __get_state():
        from . import __version__ as version

        return jsonify({"plugins": app.config["FG_PLUGINS"], "version": version})

    @app.errorhandler(Exception)
    def handle_exception(e):
        if isinstance(e, HTTPException):
            logger.debug(e.description, exc_info=True)
            return jsonify({"error": e.description}), e.code
        if isinstance(e, OperationalError):
            logger.error(e, exc_info=True)
            return {"error": "Database unavailable"}, 504
        logger.error(str(e), exc_info=True)
        return jsonify({"error": "Internal server error occurred"}), 500

    return app