#!/usr/bin/env python3
import sys, os, re, inspect, autoprop
from .layers import DictLayer, FileNotFoundLayer, dict_like
from ..utils import first_specified
from ..errors import ApiError
from pathlib import Path
from textwrap import dedent
from more_itertools import one, first
from collections.abc import Iterable
class Config:
autoload = True
dynamic = False
[docs] def __init__(self, obj, **kwargs):
self.obj = obj
self.autoload = kwargs.pop('autoload', self.autoload)
self.dynamic = kwargs.pop('dynamic', self.dynamic)
self.load_status = lambda log: None
if kwargs:
raise ApiError(
lambda e: f'{e.config.__class__.__name__}() received unexpected keyword argument(s): {", ".join(map(repr, e.kwargs))}',
config=self,
kwargs=kwargs,
)
[docs] def __repr__(self):
return f"{self.__class__.__name__}()"
[docs] @classmethod
def setup(cls, *args, **kwargs):
return lambda obj: cls(obj, *args, **kwargs)
[docs] def load(self):
raise NotImplementedError
class EnvironmentConfig(Config):
[docs] def load(self):
yield DictLayer(
values=os.environ,
location="environment",
)
class CliConfig(Config):
autoload = False
@autoprop
class ArgparseConfig(CliConfig):
parser_getter = lambda obj: obj.get_argparse()
schema = None
[docs] def __init__(self, obj, **kwargs):
self.parser_getter = kwargs.pop(
'parser_getter', unbind_method(self.parser_getter))
self.schema = kwargs.pop(
'schema', self.schema)
super().__init__(obj, **kwargs)
[docs] def load(self):
args = self.parser.parse_args()
yield DictLayer(
values=vars(args),
schema=self.schema,
location='command line',
)
[docs] def get_parser(self):
# Might make sense to cache the parser.
return self.parser_getter(self.obj)
[docs] def get_usage(self):
return self.parser.format_help()
[docs] def get_brief(self):
return self.parser.description
@autoprop
class DocoptConfig(CliConfig):
usage_getter = lambda obj: obj.__doc__
version_getter = lambda obj: getattr(obj, '__version__')
usage_io_getter = lambda obj: sys.stdout
include_help = True
include_version = None
options_first = False
schema = None
[docs] def __init__(self, obj, **kwargs):
self.usage_getter = kwargs.pop(
'usage_getter', unbind_method(self.usage_getter))
self.version_getter = kwargs.pop(
'version_getter', unbind_method(self.version_getter))
self.usage_io_getter = kwargs.pop(
'usage_io_getter', unbind_method(self.usage_io_getter))
self.include_help = kwargs.pop(
'include_help', self.include_help)
self.include_version = kwargs.pop(
'include_version', self.include_version)
self.options_first = kwargs.pop(
'options_first', self.options_first)
self.schema = kwargs.pop(
'schema', unbind_method(self.schema))
super().__init__(obj, **kwargs)
[docs] def load(self):
import sys, docopt, contextlib
with contextlib.redirect_stdout(self.usage_io):
args = docopt.docopt(
self.usage,
help=self.include_help,
version=self.version,
options_first=self.options_first,
)
# If not specified:
# - options with arguments will be None.
# - options without arguments (i.e. flags) will be False.
# - variable-number positional arguments (i.e. [<x>...]) will be []
not_specified = [None, False, []]
args = {k: v for k, v in args.items() if v not in not_specified}
yield DictLayer(
values=args,
schema=self.schema,
location='command line',
)
[docs] def get_usage(self):
from mako.template import Template
usage = self.usage_getter(self.obj)
usage = dedent(usage)
usage = Template(usage, strict_undefined=True).render(app=self.obj)
# Trailing whitespace can cause unnecessary line wrapping.
usage = re.sub(r' *$', '', usage, flags=re.MULTILINE)
return usage
[docs] def get_usage_io(self):
return self.usage_io_getter(self.obj)
[docs] def get_brief(self):
import re
sections = re.split(
'\n\n|usage:',
self.usage,
flags=re.IGNORECASE,
)
return first(sections, '').replace('\n', ' ').strip()
[docs] def get_version(self):
return self.include_version and self.version_getter(self.obj)
@autoprop
class AppDirsConfig(Config):
name = None
config_cls = None
slug = None
author = None
version = None
schema = None
root_key = None
stem = 'conf'
[docs] def __init__(self, obj, **kwargs):
self.name = kwargs.pop('name', self.name)
self.config_cls = kwargs.pop('format', self.config_cls)
self.slug = kwargs.pop('slug', self.slug)
self.author = kwargs.pop('author', self.author)
self.version = kwargs.pop('version', self.version)
self.schema = kwargs.pop('schema', unbind_method(self.schema))
self.root_key = kwargs.pop('root_key', self.root_key)
self.stem = kwargs.pop('stem', self.stem)
super().__init__(obj, **kwargs)
[docs] def load(self):
for path, config_cls in self.config_map.items():
yield from config_cls.load_from_path(
path=path, schema=self.schema, root_key=self.root_key,
)
[docs] def get_name_and_config_cls(self):
if not self.name and not self.config_cls:
raise ApiError("must specify `AppDirsConfig.name` or `AppDirsConfig.config_cls`")
if self.name and self.config_cls:
err = ApiError(
name=self.name,
format=self.config_cls,
)
err.brief = "can't specify `AppDirsConfig.name` and `AppDirsConfig.format`"
err.info += "name: {name!r}"
err.info += "format: {format!r}"
err.hints += "use `AppDirsConfig.stem` to change the filename used by `AppDirsConfig.format`"
raise err
if self.name:
suffix = Path(self.name).suffix
configs = [
x for x in FileConfig.__subclasses__()
if suffix in getattr(x, 'suffixes', ())
]
found_these = lambda e: '\n'.join([
"found these subclasses:", *(
f"{x}: {' '.join(getattr(x, 'suffixes', []))}"
for x in e.configs
)
])
with ApiError.add_info(
found_these,
name=self.name,
configs=FileConfig.__subclasses__(),
):
config = one(
configs,
ApiError("can't find FileConfig subclass to load '{name}'"),
ApiError("found multiple FileConfig subclass to load '{name}'"),
)
return self.name, config
if self.config_cls:
return self.stem + self.config_cls.suffixes[0], self.config_cls
[docs] def get_dirs(self):
from appdirs import AppDirs
slug = self.slug or self.obj.__class__.__name__.lower()
return AppDirs(slug, self.author, version=self.version)
[docs] def get_config_map(self):
dirs = self.dirs
name, config_cls = self.name_and_config_cls
return {
Path(dirs.user_config_dir) / name: config_cls,
Path(dirs.site_config_dir) / name: config_cls,
}
[docs] def get_config_paths(self):
return self.config_map.keys()
@autoprop
class FileConfig(Config):
path = None
path_getter = lambda obj: obj.path
schema = None
root_key = None
[docs] def __init__(self, obj, path=None, *, path_getter=None, schema=None, root_key=None, **kwargs):
super().__init__(obj, **kwargs)
self._path = path or self.path
self._path_getter = path_getter or unbind_method(self.path_getter)
self.schema = schema or self.schema
self.root_key = root_key or self.root_key
[docs] def get_paths(self):
try:
p = self._path or self._path_getter(self.obj)
except AttributeError as err:
def load_status(log, err=err, config=self):
log += f"failed to get path(s):\nraised {err.__class__.__name__}: {err}"
if config.paths:
br = '\n'
log += f"the following path(s) were specified post-load:{br}{br.join(str(p) for p in config.paths)}"
log += "to use these path(s), call `byoc.reload()`"
self.load_status = load_status
return []
if isinstance(p, Iterable) and not isinstance(p, str):
return [Path(pi) for pi in p]
else:
return [Path(p)]
[docs] def load(self):
for path in self.paths:
yield from self.load_from_path(
path=path,
schema=self.schema,
root_key=self.root_key,
)
[docs] @classmethod
def load_from_path(cls, path, *, schema=None, root_key=None):
try:
data, linenos = cls._do_load_with_linenos(path)
yield DictLayer(
values=data,
linenos=linenos,
location=path,
schema=schema,
root_key=root_key,
)
except FileNotFoundError:
yield FileNotFoundLayer(path)
[docs] @classmethod
def _do_load_with_linenos(cls, path):
return cls._do_load(path), {}
[docs] @staticmethod
def _do_load(path):
raise NotImplementedError
class YamlConfig(FileConfig):
suffixes = '.yml', '.yaml'
[docs] @staticmethod
def _do_load(path):
import yaml
with open(path) as f:
return yaml.safe_load(f)
class TomlConfig(FileConfig):
suffixes = '.toml',
[docs] @staticmethod
def _do_load(path):
import tomli
with open(path, 'rb') as f:
return tomli.load(f)
class NtConfig(FileConfig):
suffixes = '.nt',
[docs] @staticmethod
def _do_load_with_linenos(path):
import nestedtext as nt
keymap = {}
return nt.load(path, keymap=keymap), keymap
class JsonConfig(FileConfig):
suffixes = '.json',
[docs] @staticmethod
def _do_load(path):
import json
with open(path) as f:
return json.load(f)
def unbind_method(f):
return getattr(f, '__func__', f)