path: root/tagit/config/settings.py

"""Configuration storage.

Part of the tagit module.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# standard imports
from collections import abc
from copy import deepcopy
import io
import json
import os
import typing

# external imports
import yaml # FIXME: mb/port/convenicence

# tagit imports
from tagit.utils import errors, fst, is_list

# inner-module imports
from . import types
from .schema import schema as global_schema
from .utils import key_starts_with, superkey_of, subkey_of

# constants
INDENT = 4

# exports
__all__: typing.Sequence[str] = (
    'Settings',
    )


## code ##

class ConfigError(TypeError): pass

class Settings(abc.MutableMapping, abc.Hashable, abc.Callable):
    """Access and modify config keys in a dict-like manner.

    It's assumed that the schema might not be available for all config
    elements. That's because it might be declared in code that is
    not yet or will never be loaded. In such cases, any value is accepted.
    If the schema is known, however, it is enforced.
    """

    ## construction

    def __init__(self, schema=None, prefix=None, data=None):
        self.prefix = tuple(prefix) if prefix is not None else tuple()
        self.schema = schema if schema is not None else global_schema
        self.config = data if data is not None else dict()

    @classmethod
    def Open(cls, source, schema=None, on_error='raise', clear_defaults=True):
        schema = schema if schema is not None else global_schema
        config_path = ''
        # load source
        if isinstance(source, dict): # dictionary
            config = source
        elif isinstance(source, str): # path or serialized
            if os.path.exists(source):
                config_path = os.path.realpath(source)
                with open(source, 'r') as ifile:
                    #config = json.load(ifile)
                    config = yaml.safe_load(ifile) # FIXME: mb/port/convenicence
            else:
                #config = json.loads(source)
                config = yaml.safe_load(source) # FIXME: mb/port/convenicence
        elif isinstance(source, io.TextIOBase): # opened file
            #config = json.load(source)
            config = yaml.safe_load(source) # FIXME: mb/port/convenicence
        else:
            raise TypeError('expected dict, path, or file-like')

        # flatten and verify
        data = None
        if len(config) > 0:
            data = cls.flatten_tree(config, schema, on_error=on_error)
            if clear_defaults:
                # filter defaults
                data = {key: value
                        for key, value in data.items()
                        if key not in schema or value != schema[key].default}

        data['session', 'paths', 'config'] = config_path
        return cls(schema=schema, data=data)

    def update(self, other):
        for key, value in other.config.items():
            self.set(key, value)
        return self

    def rebase(self, schema=None, on_error='raise', clear_defaults=True):
        """Re-align the config with the current schema.
        Should be done if the schema changes *after* the Settings was initialized.
        Can also be used to enforce a new schema on the current config.

        Be aware that calling rebase will disconnect Settings instances from
        each other. For example, this affects non-leaf key retrieval via
        get such as cfg('session')
        """
        schema = self.schema if schema is None else schema
        # unroll
        tree = dict()
        for key, value in self.config.items():
            path, leaf = list(key[:-1]), key[-1]
            # navigate through the path
            branch = tree
            while len(path):
                curr = path.pop(0)
                if curr not in branch:
                    branch[curr] = dict()
                branch = branch[curr]

            branch[leaf] = value

        # flatten the unrolled config
        flat = self.flatten_tree(tree, schema, on_error=on_error)
        # remove defaults
        if clear_defaults:
            flat = {key: value
                    for key, value in flat.items()
                    if key not in schema or value != schema[key].default}
        # set new schema and config
        self.config = flat
        self.schema = schema

    ## comparison

    def __eq__(self, other):
        return isinstance(other, Settings) and \
               self.schema == other.schema and \
               self.config == other.config and \
               self.prefix == other.prefix

    def __hash__(self):
        return hash((type(self),
                     self.prefix,
                     hash(self.schema),
                     hash(tuple(sorted(self.config.items(), key=fst)))))

    def __str__(self):
        return str({k: v for k, v in self.config.items() if key_starts_with(k, self.prefix)})

    def __repr__(self):
        prefix = ','.join(self.prefix)
        size_self = len([key for key in self.config if key_starts_with(key, self.prefix)])
        size_all = len(self)
        return f'Settings(prefix=({prefix}), keys={size_self}, len={size_all})'

    ## conversion

    def clone(self):
        return Settings(schema=self.schema, prefix=self.prefix, data=deepcopy(self.config))

    @staticmethod
    def flatten_tree(hdict, schema, on_error='raise', prefix=tuple()):
        """Flattens a hierarchical dictionary by using schema information.
        Returns a flat list of config keys and their values.

        If an invalid type was found and on_error is 'raise, a TypeError is raised.
        Otherwise the invalid key is ignored.
        """
        if len(hdict) == 0:
            # not in schema, or passed the check
            return {prefix: dict()}

        flat = dict()
        for sub in hdict:
            try:
                key = prefix + (sub, )
                # check schema first, to preserve dict types
                if key in schema and schema[key].backtrack(hdict[sub]):
                        # accept the value (also defaults!)
                        flat[key] = hdict[sub]

                elif isinstance(hdict[sub], dict):
                    flat.update(Settings.flatten_tree(hdict[sub], schema, on_error, key))

                elif any(key_starts_with(k, key) for k in schema):
                    subkeys = [k[len(key):] for k in schema if key_starts_with(k, key)]
                    subkeys = ','.join('.'.join(k) for k in subkeys)
                    raise ConfigError(
                            f'found value {hdict[sub]} in {key}, expected keys ({subkeys})')

                else:
                    # terminal but not in schema; accept
                    flat[key] = hdict[sub]

            except TypeError as e:
                if on_error == 'raise':
                    raise e

        return flat

    def to_tree(self, defaults=False):
        """Return a nested dictionary with all config values.
        If *defaults*, the schema defaults are included.
        """
        tree = dict()
        source = set(self.config.keys())
        if defaults:
            source |= set(self.schema.keys())

        for key in source:
            if not key_starts_with(key, self.prefix):
                continue
            value = self.get(*key[len(self.prefix):])
            path, leaf = list(key[:-1]), key[-1]
            path = path[len(self.prefix):]
            # navigate through the path
            branch = tree
            while len(path):
                curr = path.pop(0)
                if curr not in branch:
                    branch[curr] = dict()
                branch = branch[curr]

            branch[leaf] = value

        return tree

    def file_connected(self):
        """Return True if the config is backed by a file."""
        return self('session', 'paths', 'config') is not None and \
               self('session', 'paths', 'config') != ''

    def save(self, uri=None):
        """Save changes to a file at *uri*."""
        # pick defaults
        uri = uri if uri is not None else self('session', 'paths', 'config')
        if uri is None or uri == '':
            raise ValueError('config saving requires a valid uri')

        # convert to tree
        config = self.to_tree(defaults=False)

        # save to file
        if isinstance(uri, io.TextIOBase):
            json.dump(config, uri, indent=INDENT)
        else:
            with open(uri, 'w') as ofile:
                json.dump(config, ofile, indent=INDENT)

    def diff(self, other):
        """Return a config that includes only the keys which differ from *other*."""
        # keys in self that differ from other
        config = {key: value
                  for key, value in self.config.items()
                  if key not in other.config or value != other.config[key]
                 }
        # keys in other that differ from default
        config.update({key: self.schema[key].default
                       for key, value in other.config.items()
                       if key not in self.config and \
                          key in self.schema and \
                          value != self.schema[key].default
                      })

        return Settings(schema=self.schema, prefix=self.prefix, data=deepcopy(config))


    ## getting

    def __getitem__(self, key):
        """Alias for *get*."""
        if is_list(key):
            return self.get(*key)
        else:
            return self.get(key)

    def __call__(self, *key, default=None):
        """Alias for *get*."""
        return self.get(*key, default=default)

    def get(self, *key, default=None):
        key = self.prefix + key

        # known leaf
        if key in self.config:
            value = self.config[key]
            if key in self.schema:
                if self.schema[key].check(value):
                    return value
                elif default is not None and self.schema[key].check(default):
                    return default
                else:
                    return self.schema[key].default
            else:
                return value

        # unknown leaf
        if key in self.schema:
            if default is not None and self.schema[key].check(default):
                return default
            else:
                return self.schema[key].default

        # branch
        if any(key_starts_with(sub, key) for sub in self.config):
            return Settings(schema=self.schema, prefix=key, data=self.config)
        elif any(key_starts_with(sub, key) for sub in self.schema):
            return Settings(schema=self.schema, prefix=key, data=self.config)

        if default is not None:
            return default

        raise KeyError(key)

    ## checking

    def __contains__(self, key):
        """Alias for *has*."""
        return self.has(*key)

    def has(self, *key):
        key = self.prefix + key

        if key in self.config:
            # key is a known leaf
            return True
        elif key in self.schema:
            # key is an unknown leaf
            return True
        else:
            # key might be a branch
            for sub in self.config:
                if key_starts_with(sub, key):
                    return True

            for sub in self.schema:
                if key_starts_with(sub, key):
                    return True

        return False

    ## setting

    def __setitem__(self, key, value):
        """Alias for *set*."""
        if is_list(key):
            self.set(key, value)
        else:
            self.set((key, ), value)

    def set(self, key, value):
        key = self.prefix + key

        if key in self.schema and self.schema[key].backtrack(value):
            if self.schema[key].default != value:
                self.config[key] = value
            elif key in self.config: # value is default
                # reset value to default, remove from config
                del self.config[key]
            # else: value was default but not present, ignore

        elif key in self.config:
            self.config[key] = value

        elif isinstance(value, dict) and len(value) > 0:
            # flatten value and set its items individually
            subtree = self.flatten_tree(value, self.schema, prefix=key)
            for subkey, subval in subtree.items():
                # defaults will be filtered by set
                self.set(subkey[len(self.prefix):], subval)

        elif superkey_of(key, self.schema) or subkey_of(key, self.schema):
            # schema violation  in another branch
            conflicts = {'.'.join(sub)
                         for sub in self.schema
                         if key_starts_with(sub, key) or key_starts_with(key, sub)}
            raise ConfigError(f'{key} conflicts with schema keys {",".join(conflicts)}')

        elif superkey_of(key, self.config):
            # it's allowed to overwrite dict-like config values
            # Example:
            #  having defined session.paths.preview.files = 'somewhere'
            #  it's allowed to set session.paths.preview = {}
            # It's admissible iff:
            # * the value is an empty dict
            # * no subkey is in the schema (already checked in the case above)
            if value == dict():
                self.unset(*key)
                self.config[key] = value
            else:
                conflicts = {'.'.join(sub)
                             for sub in self.schema
                             if key_starts_with(sub, key) or key_starts_with(key, sub)}
                raise ConfigError(f'{key} conflicts with config keys {",".join(conflicts)}')

        elif subkey_of(key, self.config):
            # it's allowed to overwrite dict-like config values
            # Example:
            #  having defined session.paths.preview = {}
            #  it's allowed to set session.paths.preview.files = 'somewhere'
            # It's admissible iff:
            # * the superkey is an empty dict
            # * no subkey of the superkey is in the schema
            sups = [sup for sup in self.config if key_starts_with(key, sup)]
            if len(sups) != 1:
                # there can only be one super-key
                raise errors.ProgrammingError(f'expected one superkey, found {len(sups)}')

            sup = sups[0]
            if self.config[sup] == dict() and \
               len({sub for sub in self.schema if key_starts_with(sup, sub)}) == 0:
                del self.config[sup]
                self.config[key] = value
            else:
                # already have a superkey in the config that cannot be overwritten
                conflicts = '.'.join(sup)
                raise ConfigError(f'{key} conflicts with config keys {conflicts}')

        else:
            self.config[key] = value

        return self

    ## removal

    def __delitem__(self, key):
        """Alias for *unset*."""
        if is_list(key):
            self.unset(*key)
        else:
            self.unset(key)

    def unset(self, *key):
        key = self.prefix + key

        if key in self.config:
            # key is a leaf
            del self.config[key]
        else:
            # key might be a branch
            subs = [sub for sub in self.config if key_starts_with(sub, key)]
            for sub in subs:
                del self.config[sub]

        return self

    ## iteration

    def __iter__(self):
        """Alias for *keys*."""
        return self.keys()

    def keys(self):
        for key in set(self.config.keys()) | set(self.schema.keys()):
            if key_starts_with(key, self.prefix):
                yield key[len(self.prefix):]

    def items(self):
        for key in self.keys():
            yield key, self.get(*key)

    ## properties

    def __len__(self):
        return len(list(self.keys()))


## config ##

global_schema.declare(('session', 'verbose'), types.Unsigned(), 0,
    __name__, 'Verbosity', 'Print additional information in various places of the application.')

global_schema.declare(('session', 'debug'), types.Bool(), False,
    __name__, 'Debug mode', 'Enable debug output and debug behaviour. Should be set to false in a productive environment.')

global_schema.declare(('session', 'paths', 'config'), types.Path(), '',
    __name__, 'Config path', "The path of the session's main configuration file. Is set automatically and for internal use only.")

global_schema.declare(('storage', 'config', 'write_through'), types.Bool(), True,
    __name__, 'Write-through', "Write the config to its file whenever it changes")

## EOF ##