Mercurial > repos > borgend / file revision

#
# Borgend configuration loader
#

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

#
# Defaults
#

defaults={
   # Default: backup every 6 hours (21600 seconds)
   'backup_interval': 21600,
   # Default: retry every 15 minutes if unable to connect / unfinished backup
   'retry_interval': 900,
   # borg
   '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
#

if 'borg' not in settings:
    settings['borg']=defaults['borg']
else:
    def check_and_set(cfg, field, loc, defa, fn):
        cfg[field]=fn(cfg, field, field, loc, defa[field])
        return cfg

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

    settings['borg']=check_and_set(settings['borg'], 'executable', 'borg',
                                   defaults['borg'], check_string)

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