settings_maker.py
166 lines
| 5.7 KiB
| text/x-python
|
PythonLexer
r5088 | # Copyright (C) 2010-2023 RhodeCode GmbH | |||
r4823 | # | |||
# This program is free software: you can redistribute it and/or modify | ||||
# it under the terms of the GNU Affero General Public License, version 3 | ||||
# (only), as published by the Free Software Foundation. | ||||
# | ||||
# 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 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/>. | ||||
# | ||||
# This program is dual-licensed. If you wish to learn more about the | ||||
# RhodeCode Enterprise Edition, including its added features, Support services, | ||||
# and proprietary license terms, please see https://rhodecode.com/licenses/ | ||||
import os | ||||
import textwrap | ||||
import string | ||||
import functools | ||||
import logging | ||||
import tempfile | ||||
import logging.config | ||||
r5060 | from rhodecode.lib.type_utils import str2bool, aslist | |||
r4823 | log = logging.getLogger(__name__) | |||
r5060 | ||||
r4826 | # skip keys, that are set here, so we don't double process those | |||
set_keys = { | ||||
'__file__': '' | ||||
} | ||||
r4825 | ||||
r4823 | ||||
class SettingsMaker(object): | ||||
def __init__(self, app_settings): | ||||
self.settings = app_settings | ||||
@classmethod | ||||
def _bool_func(cls, input_val): | ||||
return str2bool(input_val) | ||||
@classmethod | ||||
def _int_func(cls, input_val): | ||||
return int(input_val) | ||||
@classmethod | ||||
r5060 | def _float_func(cls, input_val): | |||
return float(input_val) | ||||
@classmethod | ||||
r4823 | def _list_func(cls, input_val, sep=','): | |||
return aslist(input_val, sep=sep) | ||||
@classmethod | ||||
def _string_func(cls, input_val, lower=True): | ||||
if lower: | ||||
input_val = input_val.lower() | ||||
return input_val | ||||
@classmethod | ||||
def _dir_func(cls, input_val, ensure_dir=False, mode=0o755): | ||||
# ensure we have our dir created | ||||
if not os.path.isdir(input_val) and ensure_dir: | ||||
os.makedirs(input_val, mode=mode) | ||||
if not os.path.isdir(input_val): | ||||
r5095 | raise Exception(f'Dir at {input_val} does not exist') | |||
r4823 | return input_val | |||
@classmethod | ||||
def _file_path_func(cls, input_val, ensure_dir=False, mode=0o755): | ||||
dirname = os.path.dirname(input_val) | ||||
cls._dir_func(dirname, ensure_dir=ensure_dir) | ||||
return input_val | ||||
@classmethod | ||||
def _key_transformator(cls, key): | ||||
return "{}_{}".format('RC'.upper(), key.upper().replace('.', '_').replace('-', '_')) | ||||
r4825 | def maybe_env_key(self, key): | |||
# now maybe we have this KEY in env, search and use the value with higher priority. | ||||
transformed_key = self._key_transformator(key) | ||||
envvar_value = os.environ.get(transformed_key) | ||||
if envvar_value: | ||||
log.debug('using `%s` key instead of `%s` key for config', transformed_key, key) | ||||
return envvar_value | ||||
def env_expand(self): | ||||
replaced = {} | ||||
for k, v in self.settings.items(): | ||||
if k not in set_keys: | ||||
envvar_value = self.maybe_env_key(k) | ||||
if envvar_value: | ||||
replaced[k] = envvar_value | ||||
set_keys[k] = envvar_value | ||||
# replace ALL keys updated | ||||
self.settings.update(replaced) | ||||
r4823 | def enable_logging(self, logging_conf=None, level='INFO', formatter='generic'): | |||
""" | ||||
Helper to enable debug on running instance | ||||
:return: | ||||
""" | ||||
if not str2bool(self.settings.get('logging.autoconfigure')): | ||||
log.info('logging configuration based on main .ini file') | ||||
return | ||||
if logging_conf is None: | ||||
logging_conf = self.settings.get('logging.logging_conf_file') or '' | ||||
if not os.path.isfile(logging_conf): | ||||
r4825 | log.error('Unable to setup logging based on %s, ' | |||
'file does not exist.... specify path using logging.logging_conf_file= config setting. ', logging_conf) | ||||
r4823 | return | |||
r5060 | with open(logging_conf, 'rt') as f: | |||
r4823 | ini_template = textwrap.dedent(f.read()) | |||
ini_template = string.Template(ini_template).safe_substitute( | ||||
RC_LOGGING_LEVEL=os.environ.get('RC_LOGGING_LEVEL', '') or level, | ||||
RC_LOGGING_FORMATTER=os.environ.get('RC_LOGGING_FORMATTER', '') or formatter | ||||
) | ||||
with tempfile.NamedTemporaryFile(prefix='rc_logging_', suffix='.ini', delete=False) as f: | ||||
log.info('Saved Temporary LOGGING config at %s', f.name) | ||||
f.write(ini_template) | ||||
logging.config.fileConfig(f.name) | ||||
os.remove(f.name) | ||||
def make_setting(self, key, default, lower=False, default_when_empty=False, parser=None): | ||||
input_val = self.settings.get(key, default) | ||||
if default_when_empty and not input_val: | ||||
# use default value when value is set in the config but it is empty | ||||
input_val = default | ||||
parser_func = { | ||||
'bool': self._bool_func, | ||||
'int': self._int_func, | ||||
r5060 | 'float': self._float_func, | |||
r4823 | 'list': self._list_func, | |||
'list:newline': functools.partial(self._list_func, sep='/n'), | ||||
'list:spacesep': functools.partial(self._list_func, sep=' '), | ||||
'string': functools.partial(self._string_func, lower=lower), | ||||
'dir': self._dir_func, | ||||
'dir:ensured': functools.partial(self._dir_func, ensure_dir=True), | ||||
'file': self._file_path_func, | ||||
'file:ensured': functools.partial(self._file_path_func, ensure_dir=True), | ||||
None: lambda i: i | ||||
}[parser] | ||||
r4825 | envvar_value = self.maybe_env_key(key) | |||
r4823 | if envvar_value: | |||
input_val = envvar_value | ||||
r4825 | set_keys[key] = input_val | |||
r4823 | self.settings[key] = parser_func(input_val) | |||
return self.settings[key] | ||||