config.py

#
# Borgend configuration loader
#

import yaml
import io
import os
import xdg
import string
import logging
from functools import reduce

#
# Defaults
#

defaults={
    # borg
    # Default: backup every 6 hours (21600 seconds)
    'backup_interval': 21600,
    # Default: retry every 15 minutes if unable to connect / unfinished backup
    'retry_interval': 900,
    # Extract passphrases at startup or on demand?
    '__extract_passphrases_at_startup': True,
    # Borg settings
    'borg': {
        'executable': 'borg',
        'common_parameters': [],
        'create_parameters': [],
        'prune_parameters': [],
    }
}


#
# Type checking etc.
#

def error(x):
    raise AssertionError(x)

def check_string(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: isinstance(x, str))

def check_dict(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: isinstance(x, dict))

def check_list(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: isinstance(x, list))

def is_list_of(x, chk):
    if x is None:
        return True
    elif isinstance(x, list):
        return reduce(lambda y, z: y and chk(z), x, True)
    else:
        return False

def check_list_of_dicts(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: is_list_of(x, lambda z: isinstance(z, dict)))

def check_list_of_strings(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: is_list_of(x, lambda z: isinstance(z, str)))

def check_nonempty_list_of_strings(cfg, field, descr, loc):
    return check_list_of_strings(cfg, field, descr, loc) and cfg[field]


def check_nonneg_int(cfg, field, descr, loc, default=None):
    return check_field(cfg, field, descr, loc, default,
                       lambda x: isinstance(x, int) and x>=0)

def check_field(cfg, field, descr, loc, default, check):
    if field in cfg:
        tmp=cfg[field]
        if not check(tmp):
            error("%s is of invalid type for %s" % (field, loc))
        return tmp
    else:
        if default is not None:
            return default
        else:
            error("%s is not configured for %s" % (field, loc))

#
# Conversion of config into command line
#

def arglistify(args):
    flatten=lambda l: [item for sublist in l for item in sublist]
    if args is None:
        return []
    else:
        return flatten([['--' + key, str(d[key])] for d in args for key in d])

#
# Load config on module load
#

def expand_env(cfg, env):
    if isinstance(cfg, dict):
        out={key: expand_env(val, env) for key, val in cfg.items()}
    elif isinstance(cfg, list):
        out=[expand_env(val, env) for val in cfg]
    elif isinstance(cfg, str):
        out=string.Template(cfg).substitute(os.environ)
    else:
        out=cfg

    return out

cfgfile=os.path.join(xdg.XDG_CONFIG_HOME, "borgend", "config.yaml")

logging.info("Reading configuration file %s" % cfgfile)

if not (os.path.exists(cfgfile) and os.path.isfile(cfgfile)):
    raise SystemExit(f'Configuration file required: {cfgfile}')

with io.open(cfgfile, 'r') as file:
    settings=expand_env(yaml.load(file), os.environ);

#
# Verify basic settings
#

def check_and_set(cfg, field, loc, defa, fn):
    cfg[field]=fn(cfg, field, field, loc, defa[field])

def check_parameters(cmd):
    check_and_set(settings['borg'], cmd+'_parameters',
                  'borg', defaults['borg'],
                   check_list_of_dicts)

check_and_set(settings, 'backup_interval', 'top-level', defaults, check_nonneg_int)
check_and_set(settings, 'retry_interval', 'top-level', defaults, check_nonneg_int)
check_and_set(settings, '__extract_passphrases_at_startup', 'top-level', defaults, check_nonneg_int)
check_and_set(settings, 'borg', 'top-level', defaults, check_dict)
# Check parameters within 'borg'
if True:
    check_and_set(settings['borg'], 'executable', 'borg',
                  defaults['borg'], check_string)

    check_parameters('common')
    check_parameters('create')
    check_parameters('prune')