mvw-dl/main.py

import configparser
import logging
import os
import sys
from rich.logging import RichHandler


# Exit codes
# 1: Config file invalid, it has no sections
# 2: Config file invalid, sections must define at least CONST.CFG_MANDATORY


class CONST(object):
    __slots__ = ()
    LOG_FORMAT = "%(message)s"
    CFG_DEFAULT_FILENAME = "config.ini"
    CFG_DEFAULT_ABS_PATH = os.path.join(os.getcwd(), CFG_DEFAULT_FILENAME)
    CFG_MANDATORY = "query"


CONST = CONST()
logging.basicConfig(
    level="NOTSET",
    format=CONST.LOG_FORMAT,
    datefmt="[%X]",
    handlers=[RichHandler(
        show_time=False if "SYSTEMD_EXEC_PID" in os.environ else True,
        rich_tracebacks=True
    )]
)
log = logging.getLogger("rich")
log.setLevel(logging.DEBUG)


class ConfigParser(configparser.ConfigParser):
    """Can get options() without defaults

    Taken from https://stackoverflow.com/a/12600066.
    """

    def options(self, section, no_defaults=False, **kwargs):
        if no_defaults:
            try:
                return list(self._sections[section].keys())
            except KeyError:
                raise configparser.NoSectionError(section)
        else:
            return super().options(section)


ini_defaults = []
internal_defaults = {
    "self_name": "mvw-dl",
    "tmp_base_dir": "/tmp/%(self_name)s",
    "state_base_dir": "/var/lib/%(self_name)s",
    "state_files_dir": "%(state_base_dir)s/state",
    "state_file_retention": "50",
    "state_file_name_prefix": "state-",
    "state_file_name_suffix": ".log",
    "mvw_endpoint": "http://localhost:8000/api/query"
}
config = ConfigParser(defaults=internal_defaults)
config.read(CONST.CFG_DEFAULT_FILENAME)


def print_section_header(header: str) -> str:
    return f"Loading config section '[{header}]' ..."


def validate_default_section(config_obj: configparser.ConfigParser()) -> None:
    log.debug(f"Loading config from file '{CONST.CFG_DEFAULT_ABS_PATH}' ...")
    if not config_obj.sections():
        log.error(f"No config sections found in '{CONST.CFG_DEFAULT_ABS_PATH}'. Exiting 1 ...")
        sys.exit(1)
    if config.defaults():
        log.debug(f"Symbol legend:\n"
                  f"* Global default from section '[{config_obj.default_section}]'\n"
                  f"~ Local option, doesn't exist in '[{config_obj.default_section}]'\n"
                  f"+ Local override of a value from '[{config_obj.default_section}]'\n"
                  f"= Local override, same value as in '[{config_obj.default_section}]'")
        log.debug(print_section_header(config_obj.default_section))
        for default in config_obj.defaults():
            ini_defaults.append({default: config_obj[config_obj.default_section][default]})
            log.debug(f"* {default} = {config_obj[config_obj.default_section][default]}")
    else:
        log.debug(f"No defaults defined")


def config_has_valid_section(config_obj: configparser.ConfigParser()) -> bool:
    has_valid_section = False
    for config_obj_section in config_obj.sections():
        if CONST.CFG_MANDATORY in config_obj.options(config_obj_section):
            has_valid_section = True
            break
    return has_valid_section


def is_default(config_key: str) -> bool:
    return any(config_key in ini_default for ini_default in ini_defaults)


def is_same_as_default(config_kv_pair: dict) -> bool:
    return config_kv_pair in ini_defaults


def validate_config_sections(config_obj: configparser.ConfigParser()) -> None:
    for section in config_obj.sections():
        log.debug(print_section_header(section))
        if CONST.CFG_MANDATORY not in config_obj.options(section, no_defaults=True):
            log.warning(f"Config section '[{section}]' does not have mandatory option '{CONST.CFG_MANDATORY}' set, "
                        f"skipping section ...")
            config_obj.remove_section(section)
        else:
            for key in config_obj.options(section, no_defaults=True):
                kv_prefix = "~"
                if is_default(key):
                    kv_prefix = "+"
                    if is_same_as_default({key: config_obj[section][key]}):
                        kv_prefix = "="
                log.debug(f"{kv_prefix} {key} = {config_obj[section][key]}")


validate_default_section(config)
if config_has_valid_section(config):
    validate_config_sections(config)
else:
    log.error(f"No valid config section found. A valid config section has at least the '{CONST.CFG_MANDATORY}' "
              f"option set. Exiting 2 ...")
    sys.exit(2)

quit()


# This is a sample Python script.

# Press Umschalt+F10 to execute it or replace it with your code.
# Press Double Shift to search everywhere for classes, files, tool windows, actions, and settings.


def print_hi(name):
    # Use a breakpoint in the code line below to debug your script.
    print(f'Hi, {name}')  # Press Strg+F8 to toggle the breakpoint.


# Press the green button in the gutter to run the script.
if __name__ == '__main__':
    print_hi('PyCharm')

# See PyCharm help at https://www.jetbrains.com/help/pycharm/