3706388589

Build Type

push

github-actions

Committed by Kale Kundert

Commit Message

feat: allow configs to control how they are matched

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

1236 of 1243 relevant lines covered (99.44%)

3.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.08

/byoc/configs/configs.py

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

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

    def __repr__(self):
        return f"{self.__class__.__name__}()"

    @classmethod
    def setup(cls, *args, **kwargs):
        return lambda obj: cls(obj, *args, **kwargs)

    def load(self):
        raise NotImplementedError

    def is_match(self, cls):
        return isinstance(self, cls)


class EnvironmentConfig(Config):

    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

    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)

    def load(self):
        args = self.parser.parse_args()
        yield DictLayer(
                values=vars(args),
                schema=self.schema,
                location='command line',
        )

    def get_parser(self):
        # Might make sense to cache the parser.
        return self.parser_getter(self.obj)

    def get_usage(self):
        return self.parser.format_help()

    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: obj.usage_io
    usage_vars_getter = lambda obj: obj.usage_vars
    include_help = True
    include_version = None
    options_first = False
    schema = None

    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.usage_vars_getter = kwargs.pop(
                'usage_vars_getter', unbind_method(self.usage_vars_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)

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

    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,
                **self.usage_vars,
        )

        # Trailing whitespace can cause unnecessary line wrapping.
        usage = re.sub(r' *$', '', usage, flags=re.MULTILINE)

        return usage

    def get_usage_io(self):
        try:
            return self.usage_io_getter(self.obj)
        except AttributeError:
            return sys.stdout

    def get_usage_vars(self):
        try:
            return self.usage_vars_getter(self.obj)
        except AttributeError:
            return {}

    def get_brief(self):
        import re
        sections = re.split(
                '\n\n|usage:',
                self.usage,
                flags=re.IGNORECASE,
        )
        return first(sections, '').replace('\n', ' ').strip()

    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'

    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)

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

    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

    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)

    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,
        }

    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

    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

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

    def load(self):
        for path in self.paths:
            yield from self.load_from_path(
                    path=path,
                    schema=self.schema,
                    root_key=self.root_key,
            )

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

    @classmethod
    def _do_load_with_linenos(cls, path):
        return cls._do_load(path), {}

    @staticmethod
    def _do_load(path):
        raise NotImplementedError

class YamlConfig(FileConfig):
    suffixes = '.yml', '.yaml'

    @staticmethod
    def _do_load(path):
        import yaml
        with open(path) as f:
            return yaml.safe_load(f)


class TomlConfig(FileConfig):
    suffixes = '.toml',

    @staticmethod
    def _do_load(path):
        import tomli
        with open(path, 'rb') as f:
            return tomli.load(f)


class NtConfig(FileConfig):
    suffixes = '.nt',

    @staticmethod
    def _do_load_with_linenos(path):
        import nestedtext as nt
        keymap = {}
        return nt.load(path, keymap=keymap), keymap

class JsonConfig(FileConfig):
    suffixes = '.json',

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

1	#!/usr/bin/env python3
2
3	import sys, os, re, inspect, autoprop	4✔
4
5	from .layers import DictLayer, FileNotFoundLayer, dict_like	4✔
6	from ..utils import first_specified	4✔
7	from ..errors import ApiError	4✔
8	from pathlib import Path	4✔
9	from textwrap import dedent	4✔
10	from more_itertools import one, first	4✔
11	from collections.abc import Iterable	4✔
12
13	class Config:	4✔
14	autoload = True	4✔
15	dynamic = False	4✔
16
17	def __init__(self, obj, **kwargs):	4✔
18	self.obj = obj	4✔
19	self.autoload = kwargs.pop('autoload', self.autoload)	4✔
20	self.dynamic = kwargs.pop('dynamic', self.dynamic)	4✔
21	self.load_status = lambda log: None	4✔
22
23	if kwargs:	4✔
24	raise ApiError(	4✔
25	lambda e: f'{e.config.__class__.__name__}() received unexpected keyword argument(s): {", ".join(map(repr, e.kwargs))}',
26	config=self,
27	kwargs=kwargs,
28	)
29
30	def __repr__(self):	4✔
31	return f"{self.__class__.__name__}()"	4✔
32
33	@classmethod	4✔
34	def setup(cls, args, *kwargs):	3✔
35	return lambda obj: cls(obj, args, *kwargs)	4✔
36
37	def load(self):	4✔
38	raise NotImplementedError	×
39
40	def is_match(self, cls):	4✔
41	return isinstance(self, cls)	4✔
42
43
44	class EnvironmentConfig(Config):	4✔
45
46	def load(self):	4✔
47	yield DictLayer(	4✔
48	values=os.environ,
49	location="environment",
50	)
51
52	class CliConfig(Config):	4✔
53	autoload = False	4✔
54
55	@autoprop	4✔
56	class ArgparseConfig(CliConfig):	4✔
57	parser_getter = lambda obj: obj.get_argparse()	4✔
58	schema = None	4✔
59
60	def __init__(self, obj, **kwargs):	4✔
61	self.parser_getter = kwargs.pop(	4✔
62	'parser_getter', unbind_method(self.parser_getter))
63	self.schema = kwargs.pop(	4✔
64	'schema', self.schema)
65
66	super().__init__(obj, **kwargs)	4✔
67
68	def load(self):	4✔
69	args = self.parser.parse_args()	4✔
70	yield DictLayer(	4✔
71	values=vars(args),
72	schema=self.schema,
73	location='command line',
74	)
75
76	def get_parser(self):	4✔
77	# Might make sense to cache the parser.
78	return self.parser_getter(self.obj)	4✔
79
80	def get_usage(self):	4✔
81	return self.parser.format_help()	4✔
82
83	def get_brief(self):	4✔
84	return self.parser.description	4✔
85
86	@autoprop	4✔
87	class DocoptConfig(CliConfig):	4✔
88	usage_getter = lambda obj: obj.__doc__	4✔
89	version_getter = lambda obj: getattr(obj, '__version__')	4✔
90	usage_io_getter = lambda obj: obj.usage_io	4✔
91	usage_vars_getter = lambda obj: obj.usage_vars	4✔
92	include_help = True	4✔
93	include_version = None	4✔
94	options_first = False	4✔
95	schema = None	4✔
96
97	def __init__(self, obj, **kwargs):	4✔
98	self.usage_getter = kwargs.pop(	4✔
99	'usage_getter', unbind_method(self.usage_getter))
100	self.version_getter = kwargs.pop(	4✔
101	'version_getter', unbind_method(self.version_getter))
102	self.usage_io_getter = kwargs.pop(	4✔
103	'usage_io_getter', unbind_method(self.usage_io_getter))
104	self.usage_vars_getter = kwargs.pop(	4✔
105	'usage_vars_getter', unbind_method(self.usage_vars_getter))
106	self.include_help = kwargs.pop(	4✔
107	'include_help', self.include_help)
108	self.include_version = kwargs.pop(	4✔
109	'include_version', self.include_version)
110	self.options_first = kwargs.pop(	4✔
111	'options_first', self.options_first)
112	self.schema = kwargs.pop(	4✔
113	'schema', unbind_method(self.schema))
114
115	super().__init__(obj, **kwargs)	4✔
116
117	def load(self):	4✔
118	import sys, docopt, contextlib	4✔
119
120	with contextlib.redirect_stdout(self.usage_io):	4✔
121	args = docopt.docopt(	4✔
122	self.usage,
123	help=self.include_help,
124	version=self.version,
125	options_first=self.options_first,
126	)
127
128	# If not specified:
129	# - options with arguments will be None.
130	# - options without arguments (i.e. flags) will be False.
131	# - variable-number positional arguments (i.e. [<x>...]) will be []
132	not_specified = [None, False, []]	4✔
133	args = {k: v for k, v in args.items() if v not in not_specified}	4✔
134
135	yield DictLayer(	4✔
136	values=args,
137	schema=self.schema,
138	location='command line',
139	)
140
141	def get_usage(self):	4✔
142	from mako.template import Template	4✔
143
144	usage = self.usage_getter(self.obj)	4✔
145	usage = dedent(usage)	4✔
146	usage = Template(usage, strict_undefined=True).render(	4✔
147	app=self.obj,
148	**self.usage_vars,
149	)
150
151	# Trailing whitespace can cause unnecessary line wrapping.
152	usage = re.sub(r' *$', '', usage, flags=re.MULTILINE)	4✔
153
154	return usage	4✔
155
156	def get_usage_io(self):	4✔
157	try:	4✔
158	return self.usage_io_getter(self.obj)	4✔
159	except AttributeError:	4✔
160	return sys.stdout	4✔
161
162	def get_usage_vars(self):	4✔
163	try:	4✔
164	return self.usage_vars_getter(self.obj)	4✔
165	except AttributeError:	4✔
166	return {}	4✔
167
168	def get_brief(self):	4✔
169	import re	4✔
170	sections = re.split(	4✔
171	'\n\n\|usage:',
172	self.usage,
173	flags=re.IGNORECASE,
174	)
175	return first(sections, '').replace('\n', ' ').strip()	4✔
176
177	def get_version(self):	4✔
178	return self.include_version and self.version_getter(self.obj)	4✔
179
180
181	@autoprop	4✔
182	class AppDirsConfig(Config):	4✔
183	name = None	4✔
184	config_cls = None	4✔
185	slug = None	4✔
186	author = None	4✔
187	version = None	4✔
188	schema = None	4✔
189	root_key = None	4✔
190	stem = 'conf'	4✔
191
192	def __init__(self, obj, **kwargs):	4✔
193	self.name = kwargs.pop('name', self.name)	4✔
194	self.config_cls = kwargs.pop('format', self.config_cls)	4✔
195	self.slug = kwargs.pop('slug', self.slug)	4✔
196	self.author = kwargs.pop('author', self.author)	4✔
197	self.version = kwargs.pop('version', self.version)	4✔
198	self.schema = kwargs.pop('schema', unbind_method(self.schema))	4✔
199	self.root_key = kwargs.pop('root_key', self.root_key)	4✔
200	self.stem = kwargs.pop('stem', self.stem)	4✔
201
202	super().__init__(obj, **kwargs)	4✔
203
204	def load(self):	4✔
205	for path, config_cls in self.config_map.items():	4✔
206	yield from config_cls.load_from_path(	4✔
207	path=path, schema=self.schema, root_key=self.root_key,
208	)
209
210	def get_name_and_config_cls(self):	4✔
211	if not self.name and not self.config_cls:	4✔
212	raise ApiError("must specify `AppDirsConfig.name` or `AppDirsConfig.config_cls`")	4✔
213
214	if self.name and self.config_cls:	4✔
215	err = ApiError(	4✔
216	name=self.name,
217	format=self.config_cls,
218	)
219	err.brief = "can't specify `AppDirsConfig.name` and `AppDirsConfig.format`"	4✔
220	err.info += "name: {name!r}"	4✔
221	err.info += "format: {format!r}"	4✔
222	err.hints += "use `AppDirsConfig.stem` to change the filename used by `AppDirsConfig.format`"	4✔
223	raise err	4✔
224
225	if self.name:	4✔
226	suffix = Path(self.name).suffix	4✔
227	configs = [	4✔
228	x for x in FileConfig.__subclasses__()
229	if suffix in getattr(x, 'suffixes', ())
230	]
231	found_these = lambda e: '\n'.join([	4✔
232	"found these subclasses:", *(
233	f"{x}: {' '.join(getattr(x, 'suffixes', []))}"
234	for x in e.configs
235	)
236	])
237	with ApiError.add_info(	4✔
238	found_these,
239	name=self.name,
240	configs=FileConfig.__subclasses__(),
241	):
242	config = one(	4✔
243	configs,
244	ApiError("can't find FileConfig subclass to load '{name}'"),
245	ApiError("found multiple FileConfig subclass to load '{name}'"),
246	)
247
248	return self.name, config	4✔
249
250	if self.config_cls:	4✔
251	return self.stem + self.config_cls.suffixes[0], self.config_cls	4✔
252
253	def get_dirs(self):	4✔
254	from appdirs import AppDirs	4✔
255	slug = self.slug or self.obj.__class__.__name__.lower()	4✔
256	return AppDirs(slug, self.author, version=self.version)	4✔
257
258	def get_config_map(self):	4✔
259	dirs = self.dirs	4✔
260	name, config_cls = self.name_and_config_cls	4✔
261	return {	4✔
262	Path(dirs.user_config_dir) / name: config_cls,
263	Path(dirs.site_config_dir) / name: config_cls,
264	}
265
266	def get_config_paths(self):	4✔
267	return self.config_map.keys()	4✔
268
269
270	@autoprop	4✔
271	class FileConfig(Config):	4✔
272	path = None	4✔
273	path_getter = lambda obj: obj.path	4✔
274	schema = None	4✔
275	root_key = None	4✔
276
277	def __init__(self, obj, path=None, , path_getter=None, schema=None, root_key=None, *kwargs):	4✔
278	super().__init__(obj, **kwargs)	4✔
279	self._path = path or self.path	4✔
280	self._path_getter = path_getter or unbind_method(self.path_getter)	4✔
281	self.schema = schema or self.schema	4✔
282	self.root_key = root_key or self.root_key	4✔
283
284	def get_paths(self):	4✔
285	try:	4✔
286	p = self._path or self._path_getter(self.obj)	4✔
287
288	except AttributeError as err:	4✔
289
290	def load_status(log, err=err, config=self):	4✔
291	log += f"failed to get path(s):\nraised {err.__class__.__name__}: {err}"	4✔
292	if config.paths:	4✔
293	br = '\n'	4✔
294	log += f"the following path(s) were specified post-load:{br}{br.join(str(p) for p in config.paths)}"	4✔
295	log += "to use these path(s), call `byoc.reload()`"	4✔
296
297	self.load_status = load_status	4✔
298	return []	4✔
299
300
301	if isinstance(p, Iterable) and not isinstance(p, str):	4✔
302	return [Path(pi) for pi in p]	4✔
303	else:
304	return [Path(p)]	4✔
305
306	def load(self):	4✔
307	for path in self.paths:	4✔
308	yield from self.load_from_path(	4✔
309	path=path,
310	schema=self.schema,
311	root_key=self.root_key,
312	)
313
314	@classmethod	4✔
315	def load_from_path(cls, path, *, schema=None, root_key=None):	4✔
316	try:	4✔
317	data, linenos = cls._do_load_with_linenos(path)	4✔
318	yield DictLayer(	4✔
319	values=data,
320	linenos=linenos,
321	location=path,
322	schema=schema,
323	root_key=root_key,
324	)
325	except FileNotFoundError:	4✔
326	yield FileNotFoundLayer(path)	4✔
327
328	@classmethod	4✔
329	def _do_load_with_linenos(cls, path):	3✔
330	return cls._do_load(path), {}	4✔
331
332	@staticmethod	4✔
333	def _do_load(path):	3✔
334	raise NotImplementedError	×
335
336	class YamlConfig(FileConfig):	4✔
337	suffixes = '.yml', '.yaml'	4✔
338
339	@staticmethod	4✔
340	def _do_load(path):	3✔
341	import yaml	4✔
342	with open(path) as f:	4✔
343	return yaml.safe_load(f)	4✔
344
345
346	class TomlConfig(FileConfig):	4✔
347	suffixes = '.toml',	4✔
348
349	@staticmethod	4✔
350	def _do_load(path):	3✔
351	import tomli	4✔
352	with open(path, 'rb') as f:	4✔
353	return tomli.load(f)	4✔
354
355
356	class NtConfig(FileConfig):	4✔
357	suffixes = '.nt',	4✔
358
359	@staticmethod	4✔
360	def _do_load_with_linenos(path):	3✔
361	import nestedtext as nt	4✔
362	keymap = {}	4✔
363	return nt.load(path, keymap=keymap), keymap	4✔
364
365	class JsonConfig(FileConfig):	4✔
366	suffixes = '.json',	4✔
367
368	@staticmethod	4✔
369	def _do_load(path):	3✔
370	import json	4✔
371	with open(path) as f:	4✔
372	return json.load(f)	4✔
373
374	def unbind_method(f):	4✔
375	return getattr(f, '__func__', f)	4✔

kalekundert / byoc / 3706388589

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous