uncloud-mravi/ucloud/settings/__init__.py

import configparser
import logging
import sys
import os

from ucloud.common.etcd_wrapper import Etcd3Wrapper

logger = logging.getLogger(__name__)


class CustomConfigParser(configparser.RawConfigParser):
    def __getitem__(self, key):
        try:
            result = super().__getitem__(key)
        except KeyError as err:
            raise KeyError(
                "Key '{}' not found in configuration. Make sure you configure ucloud.".format(
                    key
                )
            ) from err
        else:
            return result


class Settings(object):
    def __init__(self, config_key="/uncloud/config/"):
        conf_name = "ucloud.conf"
        conf_dir = os.environ.get(
            "UCLOUD_CONF_DIR", os.path.expanduser("~/ucloud/")
        )
        self.config_file = os.path.join(conf_dir, conf_name)

        self.config_parser = CustomConfigParser(allow_no_value=True)
        self.config_key = config_key

        self.read_internal_values()
        try:
            self.config_parser.read(self.config_file)
        except Exception as err:
            logger.error("%s", err)

    def get_etcd_client(self):
        args = tuple()
        try:
            kwargs = {
                "host": self.config_parser.get("etcd", "url"),
                "port": self.config_parser.get("etcd", "port"),
                "ca_cert": self.config_parser.get("etcd", "ca_cert"),
                "cert_cert": self.config_parser.get(
                    "etcd", "cert_cert"
                ),
                "cert_key": self.config_parser.get("etcd", "cert_key"),
            }
        except configparser.Error as err:
            raise configparser.Error(
                "{} in config file {}".format(
                    err.message, self.config_file
                )
            ) from err
        else:
            try:
                wrapper = Etcd3Wrapper(*args, **kwargs)
            except Exception as err:
                logger.error(
                    "etcd connection not successfull. Please check your config file."
                    "\nDetails: %s\netcd connection parameters: %s",
                    err,
                    kwargs,
                )
                sys.exit(1)
            else:
                return wrapper

    def read_internal_values(self):
        self.config_parser.read_dict(
            {
                "etcd": {
                    "file_prefix": "/files/",
                    "host_prefix": "/hosts/",
                    "image_prefix": "/images/",
                    "image_store_prefix": "/imagestore/",
                    "network_prefix": "/networks/",
                    "request_prefix": "/requests/",
                    "user_prefix": "/users/",
                    "vm_prefix": "/vms/",
                }
            }
        )

    def read_config_file_values(self, config_file):
        try:
            # Trying to read configuration file
            with open(config_file, "r") as config_file_handle:
                self.config_parser.read_file(config_file_handle)
        except FileNotFoundError:
            sys.exit(
                "Configuration file {} not found!".format(config_file)
            )
        except Exception as err:
            logger.exception(err)
            sys.exit("Error occurred while reading configuration file")

    def read_values_from_etcd(self):
        etcd_client = self.get_etcd_client()
        config_from_etcd = etcd_client.get(
            self.config_key, value_in_json=True
        )
        if config_from_etcd:
            self.config_parser.read_dict(config_from_etcd.value)
        else:
            raise KeyError(
                "Key '{}' not found in etcd. Please configure ucloud.".format(
                    self.config_key
                )
            )

    def __getitem__(self, key):
        self.read_values_from_etcd()
        return self.config_parser[key]


settings = Settings()