File size: 4,473 Bytes
2abfccb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
# -*- 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))
|