flaschengeist/flaschengeist/app.py

import enum
import json

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

from flaschengeist import logger
from flaschengeist.controller import pluginController
from flaschengeist.utils.hook import Hook
from flaschengeist.config import configure_app

from flaschengeist.database import db


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)


class CustomJSONProvider(JSONProvider):
    ensure_ascii: bool = True
    sort_keys: bool = True

    def dumps(self, obj, **kwargs):
        kwargs.setdefault("ensure_ascii", self.ensure_ascii)
        kwargs.setdefault("sort_keys", self.sort_keys)
        return json.dumps(obj, **kwargs, cls=CustomJSONEncoder)

    def loads(self, s: str | bytes, **kwargs):
        return json.loads(s, **kwargs)


@Hook("plugins.loaded")
def load_plugins(app: Flask):
    app.config["FG_PLUGINS"] = {}

    for plugin in pluginController.get_enabled_plugins():
        logger.debug(f"Searching for enabled plugin {plugin.name}")
        try:
            # Load class
            cls = plugin.entry_point.load()
            # plugin = cls.query.get(plugin.id) if plugin.id is not None else plugin
            # plugin = db.session.query(cls).get(plugin.id) if plugin.id is not None else plugin
            plugin = db.session.get(cls, plugin.id) if plugin.id is not None else plugin
            # Custom loading tasks
            plugin.load()
            # Register blueprint
            if hasattr(plugin, "blueprint") and plugin.blueprint is not None:
                app.register_blueprint(plugin.blueprint)
        except:
            logger.error(
                f"Plugin {plugin.name} was enabled, but could not be loaded due to an error.",
                exc_info=True,
            )
            continue
        logger.info(f"Loaded plugin: {plugin.name}")
        app.config["FG_PLUGINS"][plugin.name] = plugin


def create_app(test_config=None, cli=False):
    app = Flask("flaschengeist")
    app.json_provider_class = CustomJSONProvider
    app.json = CustomJSONProvider(app)
    CORS(app)

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

        configure_app(app, test_config)
        db.init_app(app)
        migrate.init_app(app, db, compare_type=True)
        load_plugins(app)

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

        return jsonify({"plugins": pluginController.get_loaded_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