config.py

#
# Aura Engine API (https://gitlab.servus.at/aura/engine-api)
#
# Copyright (C) 2020 - The Aura Engine Team.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


import logging
import os
import os.path
import sys
from configparser import ConfigParser
from pathlib import Path


class AuraConfig:
    """
    AuraConfig Class

    Holds the Aura Configuration as in the file `engine-api.ini`.
    """

    ini_path = ""
    logger = None

    def __init__(self, ini_path="/etc/aura/engine-api.ini"):
        """
        Initializes the configuration, defaults to `/etc/aura/engine.ini`.
        If this file doesn't exist it uses `./config/engine.ini` from
        the project directory.

        Args:
            ini_path(String):      The path to the configuration file `engine-api.ini`
        """
        default_ini_path = os.path.join(
            Path(__file__).parent.parent.parent.parent.absolute(),
            "config",
            "engine-api.ini",
        )

        if ini_path:
            config_file = Path(ini_path)
            if not config_file.is_file():
                ini_path = default_ini_path
        else:
            ini_path = default_ini_path

        self.ini_path = ini_path
        self.logger = logging.getLogger("AuraEngineApi")
        self.load_config()

    def set(self, key, value):
        """
        Setter for some specific config property.

        Args:
            key (String):   key
            default (*):    value
        """
        try:
            parsed_value = int(value)
        except ValueError:
            parsed_value = str(value)
        self.__dict__[key] = parsed_value

    def get(self, key, default=None):
        """
        Getter for some specific config property.

        Args:
            key (String):   key
            default (*):    value
        """
        if key not in self.__dict__:
            if default:
                self.set(key, default)
            else:
                self.logger.warning(
                    "Key " + key + " not found in configfile " + self.ini_path + "!"
                )
                return None

        if key == "debug":
            return self.__dict__[key].count("y")

        value = self.__dict__[key]
        if value and isinstance(value, str):
            value = os.path.expandvars(value)
        return value

    def load_config(self):
        """
        Set config defaults and load settings from file
        """
        if not os.path.isfile(self.ini_path):
            self.logger.critical(self.ini_path + " not found  :(")
            sys.exit(1)

        # Read the file
        f = open(self.ini_path, "r")
        ini_str = f.read()
        f.close()

        # Parse the values
        config_parser = ConfigParser()
        try:
            config_parser.read_string(ini_str)
        except Exception as e:
            self.logger.critical("Cannot read " + self.ini_path + "! Reason: " + str(e))
            sys.exit(0)

        for section in config_parser.sections():
            for key, value in config_parser.items(section):
                v = config_parser.get(section, key).replace('"', "").strip()
                self.set(key, v)

        # Custom overrides and defaults
        self.set("install_dir", os.path.realpath(__file__ + ".."))
        self.set("api_prefix", "/api/v1")

    def get_database_uri(self):
        """
        Retrieves the database connection string.
        """
        db_name = str(self.get("db_name"))
        db_type = str(self.get("db_type"))
        if db_type in {"mysql", "postgresql"}:
            db_user = str(self.get("db_user"))
            db_pass = str(self.get("db_pass"))
            db_host = str(self.get("db_host"))
            db_charset = self.get("db_charset", "utf8")
            if db_type == "mysql":
                return f"mysql://{db_user}:{db_pass}@{db_host}/{db_name}?charset={db_charset}"
            else:
                return (
                    f"postgresql+psycopg2://{db_user}:{db_pass}@{db_host}/{db_name}"
                    f"?client_encoding={db_charset}"
                )
        elif db_type == "sqlite":
            # "db_name" is expected to be either a relative or an absolute path to the sqlite file
            return f"sqlite:///{db_name}.db"
        else:
            return f"Error: invalid database type '{db_type}'"