petrel_client/common/config.py

# -*- coding: utf-8 -*-

import configparser
import logging

from petrel_client.common import exception

DEFAULT_SECTION_NAME = 'DEFAULT'


# 如果配置文件中有 [DEFAULT]，对应内容将覆盖此处
CONFIG_DEFAULT = {
    'endpoint_url': '%(host_base)s',
    'enable_mc': 'False',  # 必须为 str 类型
    'debug_mc': 'True',
    'file_log_level': 'DEBUG',
    'file_log_max_bytes': 1024 * 1024 * 1024,  # 1GB
    'file_log_backup_count': 1,
    'console_log_level': 'WARNING',
    'boto': 'True',
    'mc_server_list_path': '/mnt/lustre/share/memcached_client/server_list.conf',
    'mc_client_config_path': '/mnt/lustre/share/memcached_client/client.conf',
    'count_disp': '5000',
    'enable_mem_trace': 'False',
    'fake': 'False',
    'mc_key_cb': 'identity',
    'get_retry_max': '10',
    's3_cpp_log_level': 'off',
    # 'host_bucket': '%(host_base)s/%(bucket)s',
    # 'user_https': 'False',
    # 'ca_certs_file': '',
    # 'check_ssl_certificate': 'True',
}

_UNSET = object()


def _value_to_str(d):
    if isinstance(d, (int, bool)):
        return str(d)
    if isinstance(d, (dict,)):
        return {
            k: _value_to_str(v) for k, v in d.items()
        }
    return d


class GetterMixin(object):

    _boolean_states = {'1': True, 'yes': True, 'true': True, 'on': True,
                       '0': False, 'no': False, 'false': False, 'off': False}

    def get(self, key, default=_UNSET):
        try:
            return self[key]
        except exception.ConfigItemNotFoundError:
            if default is _UNSET:
                raise
            else:
                return default

    def has_option(self, key):
        try:
            self[key]
            return True
        except exception.ConfigItemNotFoundError:
            return False

    def get_boolean(self, key, default=_UNSET):
        v = str(self.get(key, default)).lower()

        if v not in self._boolean_states:
            raise exception.ConfigKeyTypeError('Not a boolean: ' + key)
        return self._boolean_states[v]

    def get_int(self, key, default=_UNSET):
        try:
            return int(self.get(key, default))
        except ValueError:
            raise exception.ConfigKeyTypeError('Not a integer: ' + key)

    def get_log_level(self, key, default=_UNSET):
        v = str(self.get(key, default)).upper()
        if v not in logging._nameToLevel:
            raise exception.ConfigKeyTypeError('Not a log level: ' + key)
        return logging._nameToLevel[v]


class _my_dict(configparser._default_dict):
    pass


class Config(GetterMixin):
    def __init__(self, conf_path, *args, **kwargs):
        parser = configparser.ConfigParser(CONFIG_DEFAULT)
        r = parser.read(conf_path, encoding='utf-8')
        if len(r) == 0:
            raise exception.ConfigFileNotFoundError(conf_path)
        if len(parser.sections()) == 0:
            raise exception.ConfigSectionNotFoundError()

        defaults = parser._defaults
        all_sections = parser._sections.items()
        deleteList = []
        for section, options in all_sections:
            if section.lower() != "default":
                continue
            for name, val in options.items():
                defaults[name] = val
            deleteList.append(section)
        for deleteSection in deleteList:
            parser.remove_section(deleteSection)

        self._parser = parser
        self._default = parser.items(DEFAULT_SECTION_NAME, raw=True)

    def __getitem__(self, key):
        try:
            return Section(self._parser[key])
        except KeyError as err:
            raise exception.ConfigSectionNotFoundError(*err.args)

    def update(self, other: dict):
        for k, v in other.items():
            self._parser[k].update(_value_to_str(v))

    def default(self):
        return Section(dict(self._default))

    def items(self):
        sections = self._parser.sections()
        if len(sections) == 0:
            raise exception.ConfigSectionNotFoundError()
        return [(section, self[section]) for section in sections]


class Section(GetterMixin):

    def __init__(self, conf: dict):
        # 注意 conf 中 value 取值类型均为 str
        self._conf = conf

    def __getitem__(self, key):
        try:
            return self._conf[key]
        except KeyError as err:
            raise exception.ConfigKeyNotFoundError(*err.args)

    def update(self, other):
        self._conf.update(_value_to_str(other))