flaschengeist/flaschengeist/config.py

import os
import toml
import logging.config
import collections.abc

from pathlib import Path
from werkzeug.middleware.proxy_fix import ProxyFix
from flaschengeist import _module_path, logger


# Default config:
config = {"DATABASE": {"engine": "mysql", "port": 3306}}


def update_dict(d, u):
    for k, v in u.items():
        if isinstance(v, collections.abc.Mapping):
            d[k] = update_dict(d.get(k, {}), v)
        else:
            d[k] = v
    return d


def read_configuration(test_config):
    global config
    paths = [_module_path]

    if not test_config:
        paths.append(Path.home() / ".config")
        if "FLASCHENGEIST_CONF" in os.environ:
            paths.append(Path(os.environ.get("FLASCHENGEIST_CONF")))

    for loc in paths:
        try:
            with (loc / "flaschengeist.toml").open() as source:
                logger.warning(f"Reading config file from >{loc}<")  # default root logger, goes to stderr
                update_dict(config, toml.load(source))
        except IOError:
            pass
    if test_config:
        update_dict(config, test_config)


def configure_logger(cli=False):
    global config
    # Read default config
    logger_config = toml.load(_module_path / "logging.toml")

    if "LOGGING" in config:
        # Override with user config
        update_dict(logger_config, config.get("LOGGING"))
        # Check for shortcuts
        if "level" in config["LOGGING"] or isinstance(cli, int):
            level = cli if cli and isinstance(cli, int) else config["LOGGING"]["level"]
            logger_config["loggers"]["flaschengeist"] = {"level": level}
            logger_config["handlers"]["console"]["level"] = level
            logger_config["handlers"]["file"]["level"] = level
        if cli is True or not config["LOGGING"].get("console", True):
            logger_config["handlers"]["console"]["level"] = "CRITICAL"
        if not cli and isinstance(config["LOGGING"].get("file", False), str):
            logger_config["root"]["handlers"].append("file")
            logger_config["handlers"]["file"]["filename"] = config["LOGGING"]["file"]
            Path(config["LOGGING"]["file"]).parent.mkdir(parents=True, exist_ok=True)
        else:
            del logger_config["handlers"]["file"]
    logging.config.dictConfig(logger_config)


def configure_app(app, test_config=None, cli=False):
    global config
    read_configuration(test_config)

    configure_logger(cli)

    # Always enable this builtin plugins!
    update_dict(
        config,
        {
            "auth": {"enabled": True},
            "roles": {"enabled": True},
            "users": {"enabled": True},
            "scheduler": {"enabled": True},
        },
    )

    if "secret_key" not in config["FLASCHENGEIST"]:
        logger.warning("No secret key was configured, please configure one for production systems!")
        app.config["SECRET_KEY"] = "0a657b97ef546da90b2db91862ad4e29"
    else:
        app.config["SECRET_KEY"] = config["FLASCHENGEIST"]["secret_key"]

    if test_config is not None:
        config["DATABASE"]["engine"] = "sqlite"

    if config["DATABASE"]["engine"] == "mysql":
        engine = "mysql"
        try:
            # Try mysqlclient first
            from MySQLdb import _mysql
        except ModuleNotFoundError:
            engine += "+pymysql"
        options = "?charset=utf8mb4"
    elif config["DATABASE"]["engine"] == "postgres":
        engine = "postgresql+psycopg2"
        options = "?client_encoding=utf8"
    elif config["DATABASE"]["engine"] == "sqlite":
        engine = "sqlite"
        options = ""
        host = ""
    else:
        logger.error(f"Invalid database engine configured. >{config['DATABASE']['engine']}< is unknown")
        raise Exception
    if config["DATABASE"]["engine"] in ["mysql", "postgresql"]:
        host = "{user}:{password}@{host}:{port}".format(
            user=config["DATABASE"]["user"],
            password=config["DATABASE"]["password"],
            host=config["DATABASE"]["host"],
            port=config["DATABASE"]["port"],
        )
    app.config["SQLALCHEMY_DATABASE_URI"] = f"{engine}://{host}/{config['DATABASE']['database']}{options}"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

    if "root" in config["FLASCHENGEIST"]:
        logger.debug("Setting application root to >{}<".format(config["FLASCHENGEIST"]["root"]))
        app.config["APPLICATION_ROOT"] = config["FLASCHENGEIST"]["root"]
    if config["FLASCHENGEIST"].get("proxy", False):
        logger.debug("Fixing wsgi_app for using behind a proxy server")
        app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)