cfgs.py

"""
`cfgs`: ⚙ Serializable hierarchical dataclass settings ⚙

Also:

Simple, correct handling of config, data and cache files.
Fully compliant with the XDG Base Directory Specification.
"""

from enum import Enum
from pathlib import Path
from typing import Any, Dict, Optional, Tuple, Union
import copy
import dataclasses as dc
import json
import os
import os
import sys

File = Tuple[Union[Path, str]]
_NONE = object()


class Configs:
    def diff(self, other: Any):
        assert self.__class__ is other.__class__
        result = {}
        for f in dc.fields(self):
            s, o = getattr(self, f.name), getattr(other, f.name)
            if s != o:
                if isinstance(s, Configs):
                    assert isinstance(o, Configs)
                    o = s.diff(o)
                result[f.name] = o

        return result

    def copy_from(self, **kwargs):
        for k, v in kwargs.items():
            attr = getattr(self, k)
            if isinstance(attr, Configs):
                attr.copy_from(**v)
            else:
                setattr(self, k, v)

    def load(self, *files: File):
        for f in files:
            self.copy_from(_load(f))

    def load_from_environ(
        self,
        prefix: str,
        environ: Optional[Dict] = None,
        verbose: bool = True,
    ):
        if environ is None:
            environ = os.environ
        pre = prefix.strip('_').upper() + '_'
        items = sorted(environ.items())
        items = ((k, v) for k, v in items if k.startswith(pre))

        for k, v in items:
            attr_name = k[len(pre):].lower()
            splits = list(_split_address(v, attr_name))
            if len(splits) == 1:
                parent, attr = splits[0]
                str_val = getattr(parent, attr)
                val = _string_value(k, v, str_val)
                setattr(parent, attr, val)
            elif not verbose:
                continue
            elif not splits:
                print('No configs match', k, file=sys.err)
            else:
                print('More than one config matches', k, file=sys.err)


def _split_address(parent, key):
    if key in dir(parent):
        yield parent, key
    else:
        for k in dir(parent):
            if k.startswith(key + '_'):
                k = key[len(key) + 1:]
                new_parent = getattr(parent, k)
                yield from _split_address(new_parent, k)


def _string_value(name, v, original_value):
    if original_value is None or isinstance(v, str):
        return v

    if isinstance(original_value, int):
        return int(v)

    if isinstance(original_value, float):
        return float(v)

    if isinstance(original_value, bool):
        if v.lower() in ('t', 'true'):
            return True
        if v.lower() in ('f', 'false'):
            return False
        raise ValueError(f'Cannot understand bool {name}={v}')

    if isinstance(original_value, Enum):
        return type(original_value)[v]

    return json.loads(v)


def _load(p):
    if p.suffix == '.json':
        return json.loads(p.read_text())

    if p.suffix == '.toml':
        try:
            import tomllib
            return tomllib.loads(p.read_text())
        except ImportError:
            import tomlkit
            return tomlkit.loads(p.read_text())

    if p.suffix == '.yaml':
        import yaml
        return yaml.safe_load(p.read_text())

    raise ValueError('Do not understand suffix=' + p.suffix)


_getenv = os.environ.get
_expandvars = os.path.expandvars


class App:
    """
    `cfg.App` is the main class, but it has no methods - it just holds the
    `config`, `data`, `cache` and `xdg` objects.
    """

    DEFAULT_FORMAT = 'json'
    """The default, default file format for all Apps"""

    def __init__(
        self, name, format=DEFAULT_FORMAT, read_kwds=None, write_kwds=None
    ):
        """
        Arguments:

          name: the name of the App.  `name` is used as a directory
                name so it should not contain any characters illegal in
                pathnames

          format: the format for config and data files from this App
        """

        def path(attrname):
            path = getattr(self.xdg, attrname)
            if attrname.endswith('DIRS'):
                return [os.path.join(i, self.name) for i in path.split(':')]
            return os.path.join(path, self.name)

        _check_filename(name)

        self.name = name
        """The text name of the App"""

        self.xdg = XDG()
        """A `cfg.XFG` as of when the App was constructed."""

        self.cache = Cache(path('XDG_CACHE_HOME'))
        """A `cfg.Cache` that manages cache directories"""

        if format not in FORMATS:
            raise ValueError('Unknown format', format)

        if format == 'configparser':
            self.format = ConfigparserFormat()
            """A `cfgs.Format` representing the data format."""
        else:
            self.format = Format(format, read_kwds, write_kwds)

        h, d = path('XDG_CONFIG_HOME'), path('XDG_CONFIG_DIRS')
        self.config = Directory(h, d, self.format)
        """A `cfgs.Directory` for config files"""

        h, d = path('XDG_DATA_HOME'), path('XDG_DATA_DIRS')
        self.data = Directory(h, d, self.format)
        """A `cfgs.Directory` for data files"""


class XDG:
    """
    The XDG Base Directory Spec mandates six directories for config and data
    files, caches and runtime files, with default values that can be overridden
    through environment variables.  This class takes a snapshot of these six
    directories using the current environment.
    """

    def __init__(self):
        """
        Construct the class with a snapshot of the six XDG base directories
        """

        def get(k, v):
            return _getenv(k) or _expandvars(v)

        self.XDG_CACHE_HOME = get('XDG_CACHE_HOME', '$HOME/.cache')
        """Base directory relative to which
           user-specific non-essential (cached) data should be written
        """

        self.XDG_CONFIG_DIRS = get('XDG_CONFIG_DIRS', '/etc/xdg')
        """A set of preference ordered base directories relative to which
           configuration files should be searched
        """

        self.XDG_CONFIG_HOME = get('XDG_CONFIG_HOME', '$HOME/.config')
        """Base directory relative to which user-specific
           configuration files should be written
        """

        self.XDG_DATA_DIRS = get(
            'XDG_DATA_DIRS', '/usr/local/share/:/usr/share/'
        )
        """A set of preference ordered base directories relative to which
           data files should be searched
        """

        self.XDG_DATA_HOME = get('XDG_DATA_HOME', '$HOME/.local/share')
        """Base directory relative to which user-specific
           data files should be written
        """

        self.XDG_RUNTIME_DIR = get('XDG_RUNTIME_DIR', '')
        """Base directory relative to which
           user-specific runtime files and other file objects should be placed
        """


class Directory:
    """
    An XDG directory of persistent, formatted files
    """

    def __init__(self, home, dirs, format):
        """
        Don't call this constructor directly - use either
        `cfgs.App.config` or `cfgs.App.data` instead.
        """
        self.home = home
        self.dirs = dirs
        assert not isinstance(format, str)
        self.format = format
        self.dirs.insert(0, self.home)

    def open(self, filename=None):
        """
        Open a persistent `cfg.File`.

        Arguments:
          filename: The name of the persistent file. If None,
            `filename` defaults to `cfg.App.name` plus the format suffix

          format: A string representing the file format.  If None,
             first try to guess the filename from the filename, then use
             `self.format`
        """
        if not filename:
            basename = os.path.basename(self.home)
            suffix = FORMAT_TO_SUFFIX[self.format.name]
            filename = '%s%s' % (basename, suffix)
        elif filename.startswith('/'):
            filename = filename[1:]

        return File(self.full_name(filename), self.format)

    def all_files(self, filename):
        """
        Yield all filenames matching the argument in either the home
        directory or any of the search directories
        """
        for p in self.dirs:
            full_path = os.path.join(p, filename)
            try:
                yield open(full_path) and full_path
            except IOError:
                pass

    def full_name(self, filename):
        """
        Return the full name of a file with respect to this XDG directory
        """
        return os.path.join(self.home, filename)


class File:
    """
    A formatted data or config file where you can set and get items,
    and read or write.
    """

    def __init__(self, filename, format):
        """Do not call this constructor directly but use
        `cfg.Directory.open` instead"""

        self.filename = filename
        """The full pathname to the data file"""

        self.contents = {}
        """The contents of the formatted file, read and parsed.

        This will be a `dict` for all formats except `configparser`,
        where it will be a `configparser.SafeConfigParser`.
        """

        os.makedirs(os.path.dirname(self.filename), exist_ok=True)
        self.format = format
        self.read()

    def read(self):
        """Re-read the contents from the file"""
        try:
            with open(self.filename) as fp:
                self.contents = self.format.read(fp)
        except IOError:
            self.contents = self.format.create()
        return self.contents

    def write(self):
        """Write the contents to the file"""
        with open(self.filename, 'w') as fp:
            self.format.write(self.contents, fp)

    def as_dict(self):
        """Return a deep copy of the contents as a dict"""
        return self.format.as_dict(self.contents)

    def clear(self):
        """Clear the contents without writing"""
        self.contents.clear()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.write()


class Cache:
    """
    A class that creates caches
    """

    def __init__(self, dirname):
        """Do not call this constructor - instead use `cfgs.App.cache` """

        self.dirname = dirname
        """The full path of the root directory for all cache directories"""

    def directory(self, name='cache', cache_size=0):
        """
        Return a `cfgs.CacheDirectory`

        Arguments:
          name: The relative pathname of the cache directory

          cache_size: The number of bytes allowed in the cache.
              The default of 0 means "unlimited cache size"
        """
        name = os.path.join(self.dirname, name)
        return CacheDirectory(name, cache_size)


class CacheDirectory:
    def __init__(self, dirname, cache_size):
        """Do not call this constructor - use `cfgs.Cache.directory`"""

        self.dirname = dirname
        """The full path to this cache directory"""

        self.cache_size = cache_size
        """
        The number of bytes allowed in the cache.
        0 means "unlimited cache size"
        """

        _makedirs(self.dirname)
        self.prune()

    def open(self, filename, size_guess=0, binary=False):
        """
        Open a cached file in this directory.

        If the file already exists, it is opened for read.

        Otherwise the cache is pruned and the file is opened for write.

        Arguments:
          filename: the name of the file, relative to the cache directory
          size_guess: A guess as to how large the file will be, in bytes
          binary: if True, the file is opened in binary mode

        """
        if '/' in filename:
            raise ValueError('Subdirectories are not allowed in caches')

        bin = 'b' if binary else ''

        full = os.path.join(self.dirname, filename)
        if os.path.exists(full):
            return open(full, 'r' + bin)

        self.prune(size_guess)
        return open(full, 'w' + bin)

    def prune(self, bytes_needed=0):
        """
        Prune the cache to generate at least `bytes_needed` of free space,
        if this is possible.
        """
        if not self.cache_size:
            return

        files = os.listdir(self.dirname)
        info = {f: os.stat(os.path.join(self.dirname, f)) for f in files}
        required_size = sum(s.st_size for f, s in info.items()) + bytes_needed
        if required_size <= self.cache_size:
            return

        # Delete oldest items first
        for f, s in sorted(info.items(), key=lambda x: x[1].st_mtime):
            os.remove(os.path.join(self.dirname, f))
            required_size -= s.st_size
            if required_size <= self.cache_size:
                return


def _check_filename(filename):
    # Just a heuristic - names might pass this test and still not
    # be valid i.e. CON on Windows.
    bad_chars = _BAD_CHARS.intersection(set(filename))
    if bad_chars:
        bad_chars = ''.join(sorted(bad_chars))
        raise ValueError('Invalid characters in filename: "%s"' % bad_chars)


SUFFIX_TO_FORMAT = {
    '.cfg': 'configparser',
    '.ini': 'configparser',
    '.json': 'json',
    '.toml': 'toml',
    '.yaml': 'yaml',
    '.yml': 'yaml',
}
"""
Map file suffixes to the file format - the partial inverse
to `cfgs.FORMAT_TO_SUFFIX`
"""

FORMAT_TO_SUFFIX = {
    'configparser': '.ini',
    'json': '.json',
    'toml': '.toml',
    'yaml': '.yml',
}
"""
Map file formats to file suffix - the partial inverse
to `cfgs.SUFFIX_TO_FORMAT`
"""

FORMATS = set(SUFFIX_TO_FORMAT.values())
"""A list of all formats that `cfgs` understands."""


class Format:
    def __init__(self, format, read_kwds, write_kwds):
        self.name = format
        """The name of this format"""

        self._read_kwds = read_kwds or {}
        self._write_kwds = write_kwds or {}
        self._parser = __import__(format)

    def read(self, fp):
        """Read contents from an open file in this format"""
        load = getattr(self._parser, 'safe_load', self._parser.load)
        return load(fp, **self._read_kwds)

    def write(self, contents, fp):
        """Write contents in this format to an open file"""
        dump = getattr(self._parser, 'safe_dump', self._parser.dump)
        return dump(contents, fp, **self._write_kwds)

    def create(self):
        """Return new, empty contents"""
        return {}

    def as_dict(self, contents):
        """Convert the contents to a dict"""
        return copy.deepcopy(contents)


class ConfigparserFormat(Format):
    name = 'configparser'
    """The name of the configparser format"""

    def __init__(self):
        self._parser = __import__(self.name)

    def read(self, fp):
        """Read contents from an open file in this format"""
        contents = self.create()
        contents.readfp(fp)
        return contents

    def write(self, contents, fp):
        """Write contents in this format to an open file"""
        contents.write(fp)

    def create(self):
        """Return new, empty contents"""
        return self._parser.SafeConfigParser()

    def as_dict(self, contents):
        """Convert the contents to a dict"""
        return {k: dict(v) for k, v in contents.items()}


def _makedirs(f):  # For Python 2 compatibility
    try:
        os.makedirs(f)
    except Exception:
        pass


_BAD_CHARS = set('/\\?%*:|"<>\';')