122 lines
4.1 KiB
Python
122 lines
4.1 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
|
|
)
|
|
del plugin
|
|
continue
|
|
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")
|
|
raise RuntimeError("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
|