186 lines
6.4 KiB
Python
186 lines
6.4 KiB
Python
# Copyright (C) 2010-2024 RhodeCode GmbH
|
|
#
|
|
# 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
|
|
|
|
from rhodecode.lib.type_utils import str2bool, aslist
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# skip keys, that are set here, so we don't double process those
|
|
set_keys = {"__file__": ""}
|
|
|
|
|
|
class SettingsMaker:
|
|
def __init__(self, app_settings):
|
|
self.settings = app_settings
|
|
|
|
@classmethod
|
|
def _bool_func(cls, input_val):
|
|
if isinstance(input_val, bytes):
|
|
# decode to str
|
|
input_val = input_val.decode("utf8")
|
|
return str2bool(input_val)
|
|
|
|
@classmethod
|
|
def _int_func(cls, input_val):
|
|
return int(input_val)
|
|
|
|
@classmethod
|
|
def _float_func(cls, input_val):
|
|
return float(input_val)
|
|
|
|
@classmethod
|
|
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 _string_no_quote_func(cls, input_val, lower=True):
|
|
"""
|
|
Special case string function that detects if value is set to empty quote string
|
|
e.g.
|
|
|
|
core.binar_dir = ""
|
|
"""
|
|
|
|
input_val = cls._string_func(input_val, lower=lower)
|
|
if input_val in ['""', "''"]:
|
|
return ""
|
|
|
|
@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, exist_ok=True)
|
|
|
|
if not os.path.isdir(input_val):
|
|
raise Exception(f"Dir at {input_val} does not exist")
|
|
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("-", "_"))
|
|
|
|
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):
|
|
if self.settings.get("rhodecode.env_expand") == "false":
|
|
return
|
|
|
|
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)
|
|
|
|
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):
|
|
log.error(
|
|
"Unable to setup logging based on %s, "
|
|
"file does not exist.... specify path using logging.logging_conf_file= config setting. ",
|
|
logging_conf,
|
|
)
|
|
return
|
|
|
|
with open(logging_conf, "rt") as f:
|
|
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,
|
|
"float": self._float_func,
|
|
"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),
|
|
"string:noquote": functools.partial(self._string_no_quote_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]
|
|
|
|
envvar_value = self.maybe_env_key(key)
|
|
if envvar_value:
|
|
input_val = envvar_value
|
|
set_keys[key] = input_val
|
|
|
|
self.settings[key] = parser_func(input_val)
|
|
return self.settings[key]
|