python-mylib/mylib/config.py

1446 lines
52 KiB
Python
Raw Permalink Normal View History

2023-01-06 18:16:46 +01:00
# pylint: disable=too-many-lines
2021-11-03 17:38:31 +01:00
""" Configuration & options parser """
import argparse
import logging
import os
import re
2021-11-03 17:38:31 +01:00
import sys
import textwrap
import traceback
from configparser import ConfigParser
from getpass import getpass
from logging.config import fileConfig
2021-11-03 17:38:31 +01:00
import argcomplete
import keyring
from systemd.journal import JournalHandler
log = logging.getLogger(__name__)
# Constants
DEFAULT_ENCODING = "utf-8"
DEFAULT_CONFIG_DIRPATH = os.path.expanduser("./")
DEFAULT_CONFIG_FILE_MODE = 0o600
2023-10-27 13:36:32 +02:00
DEFAULT_LOG_FORMAT = "%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s"
DEFAULT_CONSOLE_LOG_FORMAT = DEFAULT_LOG_FORMAT
DEFAULT_FILELOG_FORMAT = DEFAULT_LOG_FORMAT
2021-11-03 17:38:31 +01:00
class ConfigException(BaseException):
"""Configuration exception"""
class BaseOption: # pylint: disable=too-many-instance-attributes
"""Base configuration option class"""
def __init__(
self,
config,
section,
name,
default=None,
comment=None,
arg=None,
short_arg=None,
arg_help=None,
no_arg=False,
):
2021-11-03 17:38:31 +01:00
self.config = config
self.section = section
self.name = name
self.default = default
self.comment = comment
self.no_arg = no_arg
2021-11-03 17:38:31 +01:00
self.arg = arg
self.short_arg = short_arg
self.arg_help = arg_help if arg_help else comment
self._set = False
2021-11-03 17:38:31 +01:00
@property
def _isset_in_options(self):
"""Check if option is defined in registered arguments parser options"""
return self.config.options and not self.no_arg and self._from_options != self.default
2021-11-03 17:38:31 +01:00
@property
def _from_options(self):
"""Get option from arguments parser options"""
2021-11-03 17:38:31 +01:00
value = (
getattr(self.config.options, self.parser_dest)
if self.config.options and not self.no_arg
else None
2021-11-03 17:38:31 +01:00
)
log.debug("_from_options(%s, %s) = %s", self.section.name, self.name, value)
2021-11-03 17:38:31 +01:00
return value
@property
def _isset_in_config_file(self):
"""Check if option is defined in the loaded configuration file"""
return self.config.config_filepath and self.config.config_parser.has_option(
self.section.name, self.name
2021-11-03 17:38:31 +01:00
)
@property
def _from_config(self):
"""Get option value from ConfigParser"""
2021-11-03 17:38:31 +01:00
return self.config.config_parser.get(self.section.name, self.name)
@property
def _default_in_config(self):
"""Get option default value considering current value from configuration"""
return self._from_config if self._isset_in_config_file else self.default
2021-11-03 17:38:31 +01:00
def isset(self):
"""Check if option is defined in the loaded configuration file"""
2021-11-03 17:38:31 +01:00
return self._isset_in_config_file or self._isset_in_options
def get(self):
"""Get option value from options, config or default"""
if self._isset_in_options and not self._set:
2021-11-03 17:38:31 +01:00
return self._from_options
if self._isset_in_config_file:
return self._from_config
return self.default
def set(self, value):
"""Set option value to config file and options"""
if value == "":
2021-11-03 17:38:31 +01:00
value = None
if value == self.default or value is None:
# Remove option from config (is section exists)
if self.config.config_parser.has_section(self.section.name):
self.config.config_parser.remove_option(self.section.name, self.name)
else:
# Store option to config
if not self.config.config_parser.has_section(self.section.name):
self.config.config_parser.add_section(self.section.name)
self.config.config_parser.set(self.section.name, self.name, self.to_config(value))
self._set = True
2021-11-03 17:38:31 +01:00
def set_default(self, default_value):
"""Set option default value"""
self.default = default_value
2021-11-03 17:38:31 +01:00
@property
def parser_action(self):
"""Get action as accept by argparse.ArgumentParser"""
return "store"
2021-11-03 17:38:31 +01:00
@property
def parser_type(self):
"""Get type as handle by argparse.ArgumentParser"""
2021-11-03 17:38:31 +01:00
return str
@property
def parser_dest(self):
"""Get option name in arguments parser options"""
return f"{self.section.name}_{self.name}"
2021-11-03 17:38:31 +01:00
@property
def parser_help(self):
"""Get option help message in arguments parser options"""
2021-11-03 17:38:31 +01:00
if self.arg_help and self.default is not None:
2023-01-06 18:16:46 +01:00
# pylint: disable=consider-using-f-string
return "{} (Default: {})".format(
self.arg_help, re.sub(r"%([^%])", r"%%\1", str(self._default_in_config))
2023-01-06 18:16:46 +01:00
)
2021-11-03 17:38:31 +01:00
if self.arg_help:
return self.arg_help
return None
@property
def parser_argument_name(self):
"""Get option argument name in parser options"""
2021-11-03 17:38:31 +01:00
return (
self.arg if self.arg else f"--{self.section.name}-{self.name}".lower().replace("_", "-")
2021-11-03 17:38:31 +01:00
)
def add_option_to_parser(self, section_opts):
"""Add option to arguments parser"""
if self.no_arg:
return
2021-11-03 17:38:31 +01:00
args = [self.parser_argument_name]
if self.short_arg:
args.append(self.short_arg)
kwargs = {
"action": self.parser_action,
"dest": self.parser_dest,
"help": self.parser_help,
"default": self.default,
}
2021-11-03 17:38:31 +01:00
if self.parser_type: # pylint: disable=using-constant-test
kwargs["type"] = self.parser_type
2021-11-03 17:38:31 +01:00
log.debug(
"add_option_to_parser(%s, %s): argument name(s)=%s / kwargs=%s",
self.section.name,
self.name,
", ".join(args),
kwargs,
2021-11-03 17:38:31 +01:00
)
section_opts.add_argument(*args, **kwargs)
def to_config(self, value=None):
"""Format value as stored in configuration file"""
2021-11-03 17:38:31 +01:00
value = value if value is not None else self.get()
return "" if value is None else str(value)
2021-11-03 17:38:31 +01:00
def export_to_config(self):
"""Export option to configuration file"""
2021-11-03 17:38:31 +01:00
lines = []
if self.comment:
lines.append(f"# {self.comment}")
2021-11-03 17:38:31 +01:00
value = self.to_config()
default_value = "" if self.default is None else self.to_config(self.default)
2021-11-03 17:38:31 +01:00
log.debug(
"export_to_config(%s, %s): value=%s / default=%s",
self.section.name,
self.name,
value,
default_value,
)
2021-11-03 17:38:31 +01:00
if default_value:
if isinstance(default_value, str) and "\n" in default_value:
lines.append("# Default:")
lines.extend([f"# {line}" for line in default_value.split("\n")])
else:
lines.append(f"# Default: {default_value}")
2021-11-03 17:38:31 +01:00
if value and value != default_value:
if isinstance(value, str) and "\n" in value:
value_lines = value.split("\n")
lines.append(f"# Default: {value_lines[0]}")
lines.extend([f' {line.replace("#", "%(hash)s")}' for line in value_lines[1:]])
else:
lines.append(f"{self.name} = {value}")
2021-11-03 17:38:31 +01:00
else:
lines.append(f"# {self.name} =")
lines.append("")
return "\n".join(lines)
2021-11-03 17:38:31 +01:00
@staticmethod
def _get_user_input(prompt):
"""
Get user console input
Note: do not use directly input() to allow to mock it in tests
"""
return input(prompt)
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
if self.comment:
print(f"# {self.comment}")
default_value = kwargs.get("default_value", self.get())
if not prompt:
prompt = f"{self.name}: "
if default_value is not None:
if isinstance(default_value, str) and "\n" in default_value:
prompt += "[\n %s\n] " % "\n ".join(
self.to_config(default_value).split("\n")
)
else:
prompt += f"[{self.to_config(default_value)}] "
value = self._get_user_input(prompt)
return default_value if value == "" else value
def ask_value(self, set_it=True):
"""
Ask to user to enter value of this option and set it if set_it parameter is True
:param set_it: If True (default), option value will be updated with user input
:return: The configuration option value.
:rtype: mixed
"""
value = self._ask_value()
if set_it:
self.set(value)
return value
2021-11-03 17:38:31 +01:00
class StringOption(BaseOption):
"""String configuration option class"""
2021-11-03 17:38:31 +01:00
class BooleanOption(BaseOption):
"""Boolean configuration option class"""
2021-11-03 17:38:31 +01:00
@property
def _from_config(self):
"""Get option value from ConfigParser"""
2021-11-03 17:38:31 +01:00
return self.config.config_parser.getboolean(self.section.name, self.name)
def to_config(self, value=None):
"""Format value as stored in configuration file"""
2021-11-03 17:38:31 +01:00
return super().to_config(value).lower()
@property
def _isset_in_options(self):
"""Check if option is defined in registered arguments parser options"""
return (
self.config.options
and not self.no_arg
and getattr(self.config.options, self.parser_dest)
)
@property
def _from_options(self):
"""Get option from arguments parser options"""
return not self._default_in_config if self._isset_in_options else self._default_in_config
2021-11-03 17:38:31 +01:00
@property
def parser_action(self):
return "store_true"
2021-11-03 17:38:31 +01:00
@property
def parser_type(self):
return None
@property
def parser_argument_name(self):
"""Get option argument name in parser options"""
2023-01-06 18:16:46 +01:00
# pylint: disable=consider-using-f-string
return (
self.arg
if self.arg
else "--{}-{}-{}".format(
self.section.name, "enable" if not self._default_in_config else "disable", self.name
)
.lower()
.replace("_", "-")
)
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
default_value = self.get()
prompt = f"{self.name}: "
if default_value:
prompt += "[Y/n] "
else:
prompt += "[y/N] "
while True:
value = super()._ask_value(prompt, **kwargs)
if value in ["", None, default_value]:
return default_value
if value.lower() == "y":
return True
if value.lower() == "n":
return False
print("Invalid answer. Possible values: Y or N (case insensitive)")
2021-11-03 17:38:31 +01:00
class FloatOption(BaseOption):
"""Float configuration option class"""
2021-11-03 17:38:31 +01:00
@property
def _from_config(self):
"""Get option value from ConfigParser"""
2021-11-03 17:38:31 +01:00
return self.config.config_parser.getfloat(self.section.name, self.name)
@property
def parser_type(self):
return float
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
default_value = self.get()
while True:
value = super()._ask_value(prompt, **kwargs)
if value in ["", None, default_value]:
return default_value
try:
return float(value)
except ValueError:
print('Invalid answer. Must a numeric value, for instance "12" or "12.5"')
2021-11-03 17:38:31 +01:00
class IntegerOption(BaseOption):
"""Integer configuration option class"""
2021-11-03 17:38:31 +01:00
@property
def _from_config(self):
"""Get option value from ConfigParser"""
2021-11-03 17:38:31 +01:00
return self.config.config_parser.getint(self.section.name, self.name)
def to_config(self, value=None):
"""Format value as stored in configuration file"""
2021-11-03 17:38:31 +01:00
value = value if value is not None else self.get()
return str(int(value)) if value is not None else ""
2021-11-03 17:38:31 +01:00
@property
def parser_type(self):
return int
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
default_value = kwargs.pop("default_value", self.get())
while True:
value = super()._ask_value(prompt, default_value=default_value, **kwargs)
if value in ["", None, default_value]:
return default_value
try:
return int(value)
except ValueError:
print("Invalid answer. Must a integer value")
2021-11-03 17:38:31 +01:00
2023-12-14 17:10:34 +01:00
class OctalOption(BaseOption):
"""Octal configuration option class"""
@staticmethod
def octal(value):
"""Convert configuration octal string as integer"""
return int(str(value), 8)
@staticmethod
def octal_string(value):
"""Convert integer to configuration octal string"""
return oct(value)[2:]
@property
def _from_config(self):
"""Get option value from ConfigParser"""
return self.octal(self.config.config_parser.getint(self.section.name, self.name))
def to_config(self, value=None):
"""Format value as stored in configuration file"""
value = value if value is not None else self.get()
return self.octal_string(value) if value is not None else ""
@property
def parser_type(self):
return self.octal
@property
def parser_help(self):
"""Get option help message in arguments parser options"""
if self.arg_help and self.default is not None:
# pylint: disable=consider-using-f-string
return "{} (Default: {})".format(
self.arg_help,
re.sub(r"%([^%])", r"%%\1", self.octal_string(self._default_in_config)),
)
if self.arg_help:
return self.arg_help
return None
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
default_value = kwargs.pop("default_value", self.get())
while True:
value = super()._ask_value(prompt, default_value=default_value, **kwargs)
if value in ["", None, default_value]:
return default_value
try:
return self.octal(value)
except ValueError:
print("Invalid answer. Must an octal value")
2021-11-03 17:38:31 +01:00
class PasswordOption(StringOption):
"""Password configuration option class"""
2021-11-03 17:38:31 +01:00
def __init__(self, *arg, username_option=None, keyring_value=None, **kwargs):
super().__init__(*arg, **kwargs)
self.username_option = username_option
self.keyring_value = keyring_value if keyring_value is not None else "keyring"
2021-11-03 17:38:31 +01:00
@property
def _keyring_service_name(self):
"""Return keyring service name"""
return ".".join([self.config.shortname, self.section.name, self.name])
2021-11-03 17:38:31 +01:00
@property
def _keyring_username(self):
"""Return keyring username"""
2021-11-03 17:38:31 +01:00
return self.section.get(self.username_option) if self.username_option else self.name
def get(self):
"""Get option value"""
2021-11-03 17:38:31 +01:00
value = super().get()
if value != self.keyring_value:
return value
service_name = self._keyring_service_name
username = self._keyring_username
2024-03-15 09:52:23 +01:00
log.debug("Retrieve password %s for username=%s from keyring", service_name, username)
2021-11-03 17:38:31 +01:00
value = keyring.get_password(service_name, username)
if value is None:
2023-01-06 18:16:46 +01:00
# pylint: disable=consider-using-f-string
2021-11-03 17:38:31 +01:00
value = getpass(
"Please enter {}{}: ".format(
f"{self.section.name} {self.name}",
f" for {username}" if username != self.name else "",
2021-11-03 17:38:31 +01:00
)
)
keyring.set_password(service_name, username, value)
return value
def to_config(self, value=None):
"""Format value as stored in configuration file"""
2021-11-03 17:38:31 +01:00
if super().get() == self.keyring_value:
return self.keyring_value
return super().to_config(value)
def set(self, value, use_keyring=None): # pylint: disable=arguments-differ
"""Set option value to config file"""
if (use_keyring is None and super().get() == self.keyring_value) or use_keyring:
keyring.set_password(self._keyring_service_name, self._keyring_username, value)
2021-11-03 17:38:31 +01:00
value = self.keyring_value
super().set(value)
def _ask_value(self, prompt=None, **kwargs):
"""Ask to user to enter value of this option and return it"""
if self.comment:
print("# " + self.comment)
default_value = kwargs.pop("default_value", self.get())
if not prompt:
prompt = f"{self.name}: "
if default_value is not None:
# Hide value only if it differed from default value
if default_value == self.default:
prompt += f"[{default_value}] "
else:
prompt += "[secret defined, leave to empty to keep it as unchange] "
value = getpass(prompt)
return default_value if value == "" else value
def ask_value(self, set_it=True):
"""
Ask to user to enter value of this option and set it if set_it parameter is True
:param set_it: If True (default), option value will be updated with user input
:return: The configuration option value.
:rtype: mixed
"""
value = self._ask_value()
if set_it:
use_keyring = None
default_use_keyring = super().get() == self.keyring_value
while use_keyring is None:
prompt = (
f"Do you want to use XDG keyring ? [{'Y/n' if default_use_keyring else 'y/N'}] "
)
result = input(prompt).lower()
if result == "":
use_keyring = default_use_keyring
elif result == "y":
use_keyring = True
elif result == "n":
use_keyring = False
else:
print("Invalid answer. Possible values: Y or N (case insensitive)")
self.set(value, use_keyring=use_keyring)
return value
2021-11-03 17:38:31 +01:00
class ConfigSection:
"""Configuration section class"""
2021-11-03 17:38:31 +01:00
def __init__(self, config, name, comment=None, order=None):
self.config = config
self.name = name
2023-01-06 18:16:46 +01:00
self.options = {}
2021-11-03 17:38:31 +01:00
self.comment = comment
self.order = order if isinstance(order, int) else 10
def add_option(self, _type, name, **kwargs):
"""
Add option
:param _type: Option type, derivated from BaseOption
:param name: Option name
:param **kwargs: Dict of raw option for type class
"""
assert not self.defined(name), f"Duplicated option {name}"
2021-11-03 17:38:31 +01:00
self.options[name] = _type(self.config, self, name, **kwargs)
return self.options[name]
def defined(self, option):
"""Check if option is defined"""
2021-11-03 17:38:31 +01:00
return option in self.options
def isset(self, option):
"""Check if option is set"""
2021-11-03 17:38:31 +01:00
return self.defined(option) and self.options[option].isset()
def get(self, option):
"""Get option value"""
assert self.defined(option), f"Option {option} unknown"
2021-11-03 17:38:31 +01:00
return self.options[option].get()
def set(self, option, value):
"""Set option value"""
assert self.defined(option), f"Option {option} unknown"
2021-11-03 17:38:31 +01:00
return self.options[option].set(value)
def set_default(self, option, default_value):
"""Set default option value"""
assert self.defined(option), f"Option {option} unknown"
return self.options[option].set_default(default_value)
def set_defaults(self, **default_values):
"""Set default options value"""
for option, default_value in default_values.items():
self.set_default(option, default_value)
2021-11-03 17:38:31 +01:00
def add_options_to_parser(self, parser):
"""Add section to argparse.ArgumentParser"""
2021-11-03 17:38:31 +01:00
assert isinstance(parser, argparse.ArgumentParser)
section_opts = parser.add_argument_group(
self.comment if self.comment else self.name.capitalize()
2021-11-03 17:38:31 +01:00
)
2023-01-06 18:16:46 +01:00
for option in self.options: # pylint: disable=consider-using-dict-items
2021-11-03 17:38:31 +01:00
self.options[option].add_option_to_parser(section_opts)
def export_to_config(self):
"""Export section and their options to configuration file"""
2021-11-03 17:38:31 +01:00
lines = []
if self.comment:
lines.append(f"# {self.comment}")
lines.append(f"[{self.name}]")
2023-01-06 18:16:46 +01:00
for option in self.options: # pylint: disable=consider-using-dict-items
2021-11-03 17:38:31 +01:00
lines.append(self.options[option].export_to_config())
return "\n".join(lines)
2021-11-03 17:38:31 +01:00
def ask_values(self, set_it=True):
"""
Ask user to enter value for each configuration option of the section
:param set_it: If True (default), option value will be updated with user input
:return: a dict of configuration options and their value.
:rtype: bool of dict
"""
if self.comment:
print(f"# {self.comment}")
print(f"[{self.name}]\n")
2023-01-06 18:16:46 +01:00
result = {}
for name, option in self.options.items():
result[name] = option.ask_value(set_it=set_it)
print()
print()
return result
def ask_value(self, option, set_it=True):
"""
Ask user to enter value for the specified configuration option of the section
:param options: The configuration option name
:param set_it: If True (default), option value will be updated with user input
:return: The configuration option value.
:rtype: mixed
"""
assert self.defined(option), f"Option {option} unknown"
return self.options[option].ask_value(set_it=set_it)
2021-11-03 17:38:31 +01:00
class RawWrappedTextHelpFormatter(argparse.RawDescriptionHelpFormatter):
2021-11-23 13:08:44 +01:00
"""
Custom TextHelpFormatter for argparse.ArgumentParser that allowing line to keep line return
"""
def _split_lines(self, text, width):
result = []
for line in textwrap.dedent(text).splitlines():
# Keep empty line
if line == "":
result.append(line)
continue
# Split ident prefix and line text
m = re.match("^( *)(.*)$", line)
ident = m.group(1)
line_text = m.group(2)
# Wrap each lines and add in result with ident prefix
2022-01-20 19:14:16 +01:00
for subline in textwrap.wrap(line_text, width - len(ident)):
result.append(ident + subline)
return result
2021-11-23 13:08:44 +01:00
class Config: # pylint: disable=too-many-instance-attributes
"""Configuration helper"""
def __init__(
self,
appname,
shortname=None,
version=None,
encoding=None,
config_file_env_variable=None,
default_config_dirpath=None,
default_config_filename=None,
default_config_file_mode=None,
):
2021-11-03 17:38:31 +01:00
self.appname = appname
self.shortname = shortname
self.version = version if version else "0.0"
self.encoding = encoding if encoding else "utf-8"
2021-11-03 17:38:31 +01:00
self.config_parser = None
self.options_parser = None
self.options = None
self.sections = {}
self._loaded_callbacks = []
self._loaded_callbacks_executed = []
2021-11-03 17:38:31 +01:00
self._filepath = None
self.config_file_env_variable = config_file_env_variable
self.default_config_dirpath = default_config_dirpath
self.default_config_filename = default_config_filename
self.default_config_file_mode = default_config_file_mode or DEFAULT_CONFIG_FILE_MODE
self.add_logging_sections()
self._init_config_parser()
2021-11-03 17:38:31 +01:00
def add_logging_sections(self):
"""Add logging sections"""
console_section = self.add_section("console", comment="Console logging", order=998)
console_section.add_option(
BooleanOption,
"enabled",
default=False,
arg="--console",
short_arg="-C",
comment="Enable/disable console log",
)
console_section.add_option(
BooleanOption,
"force_stderr",
default=False,
arg="--console-stderr",
comment="Force console log on stderr",
)
console_section.add_option(
StringOption,
"log_format",
default=DEFAULT_CONSOLE_LOG_FORMAT,
arg="--console-log-format",
comment="Console log format",
)
console_section.add_option(
StringOption,
"log_level",
comment=(
"Console log level limit : by default, all logged messages (according to main log "
"level) will be logged to the console, but you can set a minimal level if you "
# logging.getLevelNamesMapping() not available in python 3.9
# pylint: disable=protected-access
f"want. Possible values: {', '.join(logging._nameToLevel)}."
),
)
logfile_section = self.add_section("logfile", comment="Logging file", order=999)
logfile_section.add_option(StringOption, "path", comment="File log path")
logfile_section.add_option(
StringOption,
"format",
default=DEFAULT_FILELOG_FORMAT,
comment="File log format",
)
logfile_section.add_option(
StringOption,
"level",
comment=(
"File log level limit : by default, all logged messages (according to main log "
"level) will be logged to the log file, but you can set a minimal level if you "
# logging.getLevelNamesMapping() not available in python 3.9
# pylint: disable=protected-access
f"want. Possible values: {', '.join(logging._nameToLevel)}."
),
)
2021-11-03 17:38:31 +01:00
def add_section(self, name, loaded_callback=None, **kwargs):
"""
Add section
: param name: The section name
: param loaded_callback: An optional callback method that will be executed after
configuration is loaded. The specified callback method will receive
Config object as parameter.
2021-11-03 17:38:31 +01:00
: param ** kwargs: Raw parameters dict pass to ConfigSection __init__() method
"""
assert name not in self.sections, f"Duplicated section {name}"
2021-11-03 17:38:31 +01:00
self.sections[name] = ConfigSection(self, name, **kwargs)
if loaded_callback:
self._loaded_callbacks.append(loaded_callback)
2024-03-15 09:52:23 +01:00
# If configuration is already loaded, execute callback immediately
if self._filepath or self.options:
self._loaded()
2021-11-03 17:38:31 +01:00
return self.sections[name]
def defined(self, section, option):
"""Check option is defined in specified section"""
2021-11-03 17:38:31 +01:00
return section in self.sections and self.sections[section].defined(option)
def isset(self, section, option):
"""Check option is set in specified section"""
2021-11-03 17:38:31 +01:00
return section in self.sections and self.sections[section].isset(option)
def get(self, section, option):
"""Get option value"""
assert self.defined(section, option), f"Unknown option {section}.{option}"
2021-11-03 17:38:31 +01:00
value = self.sections[section].get(option)
log.debug("get(%s, %s): %s (%s)", section, option, value, type(value))
2021-11-03 17:38:31 +01:00
return value
2022-05-27 15:15:34 +02:00
def get_option(self, option, default=None):
"""Get an argument parser option value"""
2022-05-27 15:15:34 +02:00
if self.options and hasattr(self.options, option):
return getattr(self.options, option)
return default
def __getitem__(self, key):
assert key in self.sections, f"Unknown section {key}"
return ConfigSectionAsDictWrapper(self.sections[key])
2021-11-03 17:38:31 +01:00
def set(self, section, option, value):
"""Set option value"""
assert self.defined(section, option), f"Unknown option {section}.{option}"
self._init_config_parser()
2021-11-03 17:38:31 +01:00
self.sections[section].set(option, value)
def set_default(self, section, option, default_value):
"""Set default option value"""
assert self.defined(section, option), f"Unknown option {section}.{option}"
self._init_config_parser()
self.sections[section].set_default(option, default_value)
def set_defaults(self, section, **default_values):
"""Set default options value"""
assert section in self.sections, f"Unknown section {section}"
self._init_config_parser()
self.sections[section].set_defaults(**default_values)
def _init_config_parser(self, force=False):
"""Initialize ConfigParser object"""
if not self.config_parser or force:
self.config_parser = ConfigParser(defaults={"hash": "#"})
def load_file(self, filepath, execute_callback=True):
"""Read configuration file"""
self._init_config_parser(force=True)
2021-11-03 17:38:31 +01:00
self._filepath = filepath
# Checking access of configuration file
if not os.path.isfile(filepath):
return True
if not os.access(filepath, os.R_OK):
return False
try:
self.config_parser.read(filepath, encoding=self.encoding)
except Exception: # pylint: disable=broad-except
self._init_config_parser(force=True)
log.exception("Failed to read configuration file %s", filepath)
2021-11-03 17:38:31 +01:00
return False
# Logging initialization
if self.config_parser.has_section("loggers"):
2021-11-03 17:38:31 +01:00
fileConfig(filepath)
else:
# Otherwise, use systemd journal handler
handler = JournalHandler(SYSLOG_IDENTIFIER=self.shortname)
handler.setFormatter(logging.Formatter("%(levelname)s | %(name)s | %(message)s"))
2021-11-03 17:38:31 +01:00
logging.getLogger().addHandler(handler)
self._filepath = filepath
if execute_callback:
self._loaded()
2021-11-03 17:38:31 +01:00
return True
def _loaded(self):
"""Execute loaded callbacks"""
error = False
for callback in self._loaded_callbacks:
if callback in self._loaded_callbacks_executed:
continue
if not callback(self):
error = True
self._loaded_callbacks_executed.append(callback)
return not error
def save(self, filepath=None, reload=True):
"""Save configuration file"""
2021-11-03 17:38:31 +01:00
filepath = filepath if filepath else self._filepath
assert filepath, "Configuration filepath is not set or provided"
2021-11-03 17:38:31 +01:00
# Checking access of target file/directory
2021-11-03 17:38:31 +01:00
dirpath = os.path.dirname(filepath)
if os.path.isfile(filepath):
if not os.access(filepath, os.W_OK):
2022-01-20 19:14:16 +01:00
log.error('Configuration file "%s" is not writable', filepath)
return False
elif not os.path.isdir(dirpath) or not os.access(dirpath, os.R_OK | os.W_OK | os.X_OK):
log.error('Configuration directory "%s" does not exist (or not writable)', dirpath)
2021-11-03 17:38:31 +01:00
return False
lines = [f"#\n# {self.appname} configuration\n#\n"]
2021-11-03 17:38:31 +01:00
for section_name in self._ordered_section_names:
lines.append("")
2021-11-03 17:38:31 +01:00
lines.append(self.sections[section_name].export_to_config())
try:
with open(filepath, "wb") as fd:
fd.write("\n".join(lines).encode(self.encoding))
2021-11-03 17:38:31 +01:00
# Privacy!
os.chmod(filepath, self.default_config_file_mode)
2021-11-03 17:38:31 +01:00
except Exception: # pylint: disable=broad-except
log.exception("Failed to write generated configuration file %s", filepath)
2021-11-03 17:38:31 +01:00
return False
if reload:
return self.load_file(filepath)
2021-11-03 17:38:31 +01:00
return True
@property
def _ordered_section_names(self):
"""Get ordered list of section names"""
2021-11-03 17:38:31 +01:00
return sorted(self.sections.keys(), key=lambda section: self.sections[section].order)
def get_arguments_parser(self, reconfigure=False, **kwargs):
"""Get arguments parser"""
2021-11-03 17:38:31 +01:00
if self.options_parser:
return self.options_parser
self.options_parser = argparse.ArgumentParser(
description=kwargs.pop("description", self.appname),
formatter_class=RawWrappedTextHelpFormatter,
**kwargs,
2023-01-06 18:16:46 +01:00
)
config_file_help = f"Configuration file to use (default: {self.config_filepath})"
if self.config_file_env_variable:
2023-01-06 18:16:46 +01:00
config_file_help += (
"\n\nYou also could set "
f"{self.config_file_env_variable} environment variable to "
"specify your configuration file path."
2023-01-06 18:16:46 +01:00
)
2021-11-03 17:38:31 +01:00
self.options_parser.add_argument(
"-c", "--config", default=self.config_filepath, help=config_file_help
2021-11-03 17:38:31 +01:00
)
self.options_parser.add_argument(
"--save",
action="store_true",
dest="save",
help="Save current configuration to file",
2021-11-03 17:38:31 +01:00
)
if reconfigure:
self.options_parser.add_argument(
"--reconfigure",
action="store_true",
dest="mylib_config_reconfigure",
help="Reconfigure and update configuration file",
)
2021-11-03 17:38:31 +01:00
self.options_parser.add_argument(
"-d", "--debug", action="store_true", help="Show debug messages"
2021-11-03 17:38:31 +01:00
)
self.options_parser.add_argument(
"-v", "--verbose", action="store_true", help="Show verbose messages"
2021-11-03 17:38:31 +01:00
)
self.add_options_to_parser(self.options_parser)
return self.options_parser
def parse_arguments_options(
self,
argv=None,
parser=None,
create=True,
ask_values=True,
exit_after_created=True,
execute_callback=True,
hardcoded_options=None,
):
2021-11-03 17:38:31 +01:00
"""
Parse arguments options
:param argv: Optional arguments list to parse (default: sys.argv[1:])
:param parser: Optional argparse.ArgumentParser use
(default: generated by self.get_arguments_parser())
:param create: If True, configuration file will be created if it does not exits
(default: True)
:param ask_values: If True, ask user to enter valor of each configuration options
(default: True)
2021-11-03 17:38:31 +01:00
:param exit_after_created: If True, script will end after configuration file
creation (default: True)
:param execute_callback: Sections's loaded callbacks will be executed only if True
(default: True)
:param hardcoded_options: Optional hard-coded options to set after loading arguments
and configuration file. These options are passed using an
list of tuple of 3 elements: the section and the option
names and the value.
[('section1', 'option1', value), ...]
2021-11-03 17:38:31 +01:00
"""
parser = parser if parser else self.get_arguments_parser()
argcomplete.autocomplete(parser)
options = parser.parse_args(argv if argv is not None else sys.argv[1:])
self.load_options(options, execute_callback=False)
2021-11-03 17:38:31 +01:00
if options.config:
options.config = os.path.abspath(options.config)
already_saved = False
if not os.path.isfile(options.config) and (create or options.save):
log.warning("Configuration file is missing, generate it (%s)", options.config)
if ask_values:
self.ask_values(set_it=True)
2021-11-03 17:38:31 +01:00
self.save(options.config)
if exit_after_created:
sys.exit(0)
already_saved = True
# Load configuration file
if os.path.isfile(options.config) and not self.load_file(
options.config, execute_callback=False
):
parser.error(f"Failed to load configuration from file {options.config}")
2021-11-03 17:38:31 +01:00
if options.save and not already_saved:
self.save()
sys.exit(0)
if options.debug:
logging.getLogger().setLevel(logging.DEBUG)
elif options.verbose:
logging.getLogger().setLevel(logging.INFO)
if hardcoded_options:
assert isinstance(hardcoded_options, list), (
"hardcoded_options must be a list of tuple of 3 elements: "
"the section and the option names and the value."
)
for opt_info in hardcoded_options:
assert isinstance(opt_info, tuple) and len(opt_info) == 3, (
"Invalid hard-coded option value: it must be a tuple of 3 "
"elements: the section and the option names and the value."
)
self.set(*opt_info)
if self.get("console", "enabled"):
console_log_level = (
# logging.getLevelNamesMapping() not available in python 3.9
# pylint: disable=protected-access
logging._nameToLevel.get(self.get("console", "log_level"))
if self.get("console", "log_level")
else logging.DEBUG
)
if console_log_level < logging.WARNING:
stdout_console_handler = logging.StreamHandler(
sys.stderr if self.get("console", "force_stderr") else sys.stdout
)
stdout_console_handler.addFilter(StdoutInfoFilter())
stdout_console_handler.setLevel(console_log_level)
stderr_console_handler = logging.StreamHandler(sys.stderr)
stderr_console_handler.setLevel(
console_log_level if console_log_level > logging.WARNING else logging.WARNING
)
if self.get("console", "log_format"):
console_formater = logging.Formatter(self.get("console", "log_format"))
if console_log_level < logging.WARNING:
stdout_console_handler.setFormatter(console_formater)
stderr_console_handler.setFormatter(console_formater)
if console_log_level < logging.WARNING:
logging.getLogger().addHandler(stdout_console_handler)
logging.getLogger().addHandler(stderr_console_handler)
2021-11-18 19:55:44 +01:00
2023-10-27 13:36:32 +02:00
if self.get("logfile", "path"):
logfile_handler = logging.FileHandler(self.get("logfile", "path"))
logfile_level = (
2023-10-27 13:42:42 +02:00
# logging.getLevelNamesMapping() not available in python 3.9
# pylint: disable=protected-access
logging._nameToLevel.get(self.get("logfile", "level"))
2023-10-27 13:36:32 +02:00
if self.get("logfile", "level")
else logging.DEBUG
)
if logfile_level is None:
log.fatal("Invalid log file level specified (%s)", self.get("logfile", "level"))
sys.exit(1)
logfile_handler.setLevel(logfile_level)
if self.get("logfile", "format"):
logfile_formater = logging.Formatter(self.get("logfile", "format"))
logfile_handler.setFormatter(logfile_formater)
logging.getLogger().addHandler(logfile_handler)
if execute_callback:
self._loaded()
if self.get_option("mylib_config_reconfigure", default=False):
self.ask_values(set_it=True)
sys.exit(1)
2021-11-03 17:38:31 +01:00
return options
def load_options(self, options, execute_callback=True):
"""Register arguments parser options"""
2021-11-03 17:38:31 +01:00
assert isinstance(options, argparse.Namespace)
self.options = options
log.debug("Argument options: %s", options)
if execute_callback:
self._loaded()
2021-11-03 17:38:31 +01:00
def add_options_to_parser(self, parser):
"""Add sections and their options to parser"""
2021-11-03 17:38:31 +01:00
for section in self._ordered_section_names:
self.sections[section].add_options_to_parser(parser)
def ask_values(self, set_it=True, execute_callback=False):
"""
Ask user to enter value for each configuration option
:param set_it: If True (default), option value will be updated with user input
:param execute_callback: Sections's loaded callbacks will be finally executed
(only if set_it is True, default: False)
:return: a dict of configuration section and their options value.
:rtype: dict
"""
2023-01-06 18:16:46 +01:00
result = {}
for name, section in self.sections.items():
result[name] = section.ask_values(set_it=set_it)
if set_it and execute_callback:
self._loaded()
return result
def ask_value(self, section, option, set_it=True):
"""
Ask user to enter value for the specified configuration option
:param section: The configuration section name
:param option: The configuration option name
:param set_it: If True (default), option value will be updated with user input
:return: The configuration option value.
:rtype: mixed
"""
assert self.defined(section, option), f"Unknown option {section}.{option}"
return self.sections[section].ask_value(option, set_it=set_it)
def configure(self, argv=None, description=False):
"""
Entry point of a script you could use to created your configuration file
Note: make sure to load to define all your configuration section and options
before running it.
"""
parser = self.get_arguments_parser(
description=description if description else "Generate configuration file"
)
parser.add_argument(
"-i",
"--interactive",
action="store_true",
dest="interactive",
help="Enable configuration interactive mode",
)
parser.add_argument(
"-O",
"--overwrite",
action="store_true",
dest="overwrite",
help="Overwrite configuration file if exists",
)
parser.add_argument(
"-V",
"--validate",
action="store_true",
dest="validate",
help=(
"Validate configuration: initialize application to test if provided parameters"
2024-03-15 09:52:23 +01:00
" works.\n\nNote: Validation will occurred after configuration file creation or"
" update. On error, re-run with -O/--overwrite parameter to fix it."
),
)
options = self.parse_arguments_options(argv, create=False, execute_callback=False)
def validate():
"""Validate configuration file"""
print("Validate your configuration...")
try:
if self.load_file(options.config):
print("Your configuration seem valid.")
else:
print("Error(s) occurred validating your configuration. See logs for details.")
sys.exit(1)
except Exception: # pylint: disable=broad-except
print(
"Exception occurred validating your configuration:\n"
f"{traceback.format_exc()}"
"\n\nSee logs for details."
)
sys.exit(2)
if os.path.exists(options.config) and not options.overwrite:
print(
f"Configuration file {options.config} already exists. "
"Use -O/--overwrite parameter to overwrite it."
)
if options.validate:
validate()
sys.exit(0)
sys.exit(1)
if options.interactive:
self.ask_values(set_it=True)
if self.save(options.config, reload=False):
print(f"Configuration file {options.config} created.")
if options.validate:
validate()
else:
2024-03-15 09:52:23 +01:00
print(f"Error occurred creating configuration file {options.config}")
sys.exit(1)
sys.exit(0)
2021-11-03 17:38:31 +01:00
@property
def config_dir(self):
"""Retrieve configuration directory path"""
if self._filepath:
return os.path.dirname(self._filepath)
if self.default_config_dirpath:
return self.default_config_dirpath
return DEFAULT_CONFIG_DIRPATH
2021-11-03 17:38:31 +01:00
@property
def config_filepath(self):
"""Retrieve configuration file path"""
2021-11-03 17:38:31 +01:00
if self._filepath:
return self._filepath
if self.config_file_env_variable and os.environ.get(self.config_file_env_variable):
return os.environ.get(self.config_file_env_variable)
if self.default_config_filename:
filename = self.default_config_filename
else:
filename = f"{self.shortname}.ini" if self.shortname else "config.ini"
return os.path.join(self.config_dir, filename)
class ConfigurableObject:
"""
Base class of configurable object
This class provide a way to configure an object using :
- mylib.config.Config object
- argparse.Namespace object
- kwargs passed to __init__ method
"""
# Configuration object name (used for default options prefix and config section)
# Note: required if options_prefix or/and config_section parameters not provided
# to __init__ method.
_config_name = None
# Configuration comment (used for config section)
_config_comment = None
# Default options value
# Important: all supported options MUST HAVE a default value defined
_defaults = {
"just_try": None,
}
# Store options passed throuht __init__ method
_kwargs = {}
_options = {}
_options_prefix = None
_config = None
_config_section = None
def __init__(
self, options=None, options_prefix=None, config=None, config_section=None, **kwargs
):
for key, value in kwargs.items():
assert key in self._defaults, f"Unknown {key} option"
self._kwargs[key] = value
if options:
self._options = options
if options_prefix is not None:
self._options_prefix = options_prefix
elif self._config_name:
self._options_prefix = self._config_name + "_"
else:
raise ConfigException(f"No configuration name defined for {__name__}")
if config:
self._config = config
if config_section:
self._config_section = config_section
elif self._config_name:
self._config_section = self._config_name
else:
raise ConfigException(f"No configuration name defined for {__name__}")
def _get_option(self, option, default=None, required=False):
2024-03-15 09:52:23 +01:00
"""Retrieve option value"""
if self._kwargs and option in self._kwargs:
return self._kwargs[option]
if self._options and hasattr(self._options, self._options_prefix + option):
return getattr(self._options, self._options_prefix + option)
if self._config and self._config.defined(self._config_section, option):
return self._config.get(self._config_section, option)
assert not required, f"Options {option} not defined"
return default if default is not None else self._defaults.get(option)
def _set_option(self, option, value):
"""Set option value"""
self._kwargs[option] = value
def set_default(self, option, default_value):
"""Set option default value"""
2024-03-15 09:52:23 +01:00
assert option in self._defaults, f"Unknown option {option}"
self._defaults[option] = default_value
def set_defaults(self, **default_values):
"""Set options default value"""
for option, default_value in default_values.items():
self.set_default(option, default_value)
def configure(
self,
comment=None,
just_try=False,
just_try_default=False,
just_try_help="Just-try mode",
**kwargs,
):
"""
Configure options on registered mylib.Config object
:param comment: Configuration section comment (default: self._config_comment)
:param just_try: Add just-try mode option (default: False)
:param just_try_default: Default just-try mode option value (default: False)
:param just_try_help: Default just-try mode option help message (default: "Just-try mode")
:param kwargs: Other provided parameters are directly passed to Config.add_section() method
"""
assert self._config, (
"mylib.Config object not registered. Must be passed to __init__ as config keyword"
" argument."
)
section = self._config.add_section(
self._config_section,
comment=comment if comment else self._config_comment,
loaded_callback=self.initialize,
**kwargs,
)
if just_try:
self._defaults["just_try"] = just_try_default
section.add_option(
BooleanOption,
"just_try",
default=self._defaults["just_try"],
comment=just_try_help if just_try_help else "Just-try mode",
)
return section
def initialize(self, loaded_config=None):
"""Configuration initialized hook"""
if loaded_config:
self.config = loaded_config # pylint: disable=attribute-defined-outside-init
@property
def _just_try(self):
"""Check if just-try mode is enabled"""
# If "just_try" provided to constructor, use it value
if "just_try" in self._kwargs:
log.debug(
"Just-try mode is %s by value passed to constructor",
"enabled" if self._kwargs["just_try"] else "disabled",
)
return self._kwargs["just_try"]
# If options provided and just-try option exist and is enabled, just-try mode enabled
if (
self._options
and hasattr(self._options, f"{self._options_prefix}just_try")
and getattr(self._options, f"{self._options_prefix}just_try")
):
log.debug("Just-try mode for %s is enabled", __class__.__name__)
return True
# If options provided and a just_try option exist and is enabled, just-try mode enabled
if (
self._options
and hasattr(self._options, "just_try")
and getattr(self._options, "just_try")
):
log.debug("Just-try mode is globally enabled")
return True
# If Config provided, config section defined and just-try enabled in config, just-try mode
# enabled
if (
self._config
and self._config.defined(self._config_section, "just_try")
and self._config.get(self._config_section, "just_try")
):
log.debug("Just-try mode for %s is enabled in configuration", self._config_section)
return True
# If Config provided, use it's get_option() method to obtain a global just_try parameter
2024-03-15 09:52:23 +01:00
# value with a default to False, otherwise always false
return self._config.get_option("just_try", default=False) if self._config else False
class ConfigSectionAsDictWrapper:
"""
Wrapper for ConfigSection that offer __getitems__ and __setitem__ methods
to allow access to section options as with a dictionary.
"""
__section = None
def __init__(self, section):
self.__section = section
def __getitem__(self, key):
return self.__section.get(key)
def __setitem__(self, key, value):
self.__section.set(key, value)
def __delitem__(self, key):
raise ConfigException("Deleting a configuration option is not supported")
# pylint: disable=too-few-public-methods
class StdoutInfoFilter(logging.Filter):
"""Logging filter to keep messages only >= logging.INFO"""
def filter(self, record):
return record.levelno in (logging.DEBUG, logging.INFO)