config.py

#!/usr/local/bin/python3
#
# Borgend configuration loader
#

import yaml
import io
import os
import string
import logging
import platform
from functools import reduce
import borgend

logger=borgend.logger.getChild(__name__)

#
# 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,
    # Do not insert a quit menu entry (useful for installing on computers of
    # inexperienced users)
    'no_quit_menu_entry': False,
    # Borg settings
    'borg': {
        'executable': 'borg',
        'common_parameters': [],
        'create_parameters': [],
        'prune_parameters': [],
    }
}

#
# Locations
#

if platform.system()!='Darwin':
    import xdg
    cfgfile=os.path.join(xdg.XDG_CONFIG_HOME, borgend.appname, "config.yaml")
    logs_dir=os.path.join(xdg.XDG_DATA_HOME, borgend.appname, "logs")
else:
    import rumps
    __base=rumps.application_support(borgend.appname)
    cfgfile=os.path.join(__base, "config.yaml")
    logs_dir=os.path.join(__base, "logs")

#
# Type checking etc.
#

def error(x):
    raise AssertionError(x)

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))

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

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)


#
# Borg command line parameter configuration helper routines and classes
#

class BorgParameters:
    def __init__(self, common, create, prune):
        self.common=common or []
        self.create=create or []
        self.prune=prune or []

    def from_config(cfg, loc):
        common=check_list_of_dicts(cfg, 'common_parameters',
                                   'Borg parameters', loc, default=[])

        create=check_list_of_dicts(cfg, 'create_parameters',
                                   'Create parameters', loc, default=[])

        prune=check_list_of_dicts(cfg, 'prune_parameters',
                                  'Prune parameters', loc, default=[])

        return BorgParameters(common, create, prune)

    def __add__(self, other):
        common=self.common+other.common
        create=self.create+other.create
        prune=self.prune+other.prune
        return BorgParameters(common, create, prune)

#
# 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

if not (os.path.exists(cfgfile) and os.path.isfile(cfgfile)):
    raise SystemExit("Configuration file required: %s" % cfgfile)

logger.info("Reading configuration %s missing" % 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])

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, 'no_quit_menu_entry', 'top-level', defaults, check_bool)
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)

    borg_parameters=BorgParameters.from_config(settings['borg'], "top-level")