flaschengeist/flaschengeist/app.py

119 lines
3.9 KiB
Python

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
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)
def __load_plugins(app):
logger.info("Search for plugins")
app.config["FG_PLUGINS"] = {}
for entry_point in pkg_resources.iter_entry_points("flaschengeist.plugin"):
logger.debug("Found plugin: >{}<".format(entry_point.name))
plugin = None
if entry_point.name in config and config[entry_point.name].get("enabled", False):
try:
logger.info(f"Load plugin {entry_point.name}")
plugin = entry_point.load()
if not hasattr(plugin, "name"):
setattr(plugin, "name", entry_point.name)
plugin = plugin(config[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
)
if isinstance(plugin, AuthPlugin):
logger.debug(f"Found authentication plugin: {entry_point.name}")
if entry_point.name == config["FLASCHENGEIST"]["auth"]:
app.config["FG_AUTH_BACKEND"] = plugin
else:
del plugin
continue
if plugin:
app.config["FG_PLUGINS"][entry_point.name] = plugin
if "FG_AUTH_BACKEND" not in app.config:
logger.error("No authentication plugin configured or authentication plugin not found")
def install_all():
from flaschengeist.database import db
db.create_all()
db.session.commit()
installed = []
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()
installed.append(plugin)
if plugin.permissions:
roleController.create_permissions(plugin.permissions)
for plugin in installed:
plugin.post_install()
def create_app(test_config=None):
app = Flask(__name__)
app.json_encoder = CustomJSONEncoder
CORS(app)
with app.app_context():
from flaschengeist.database import db
configure_app(app, test_config)
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