-
Notifications
You must be signed in to change notification settings - Fork 0
/
click_configfile.py
499 lines (413 loc) · 18.5 KB
/
click_configfile.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
# -*- coding: UTF-8 -*-
# (c) 2016 by Jens Engel
# pylint: disable=missing-docstring, too-few-public-methods
"""
This module provides functionality to read configuration files with commands
that already use click_ for command-line processing.
.. _click: http://click.pocoo.org/
"""
from __future__ import absolute_import, print_function
from fnmatch import fnmatch
import os.path
import inspect
import configparser # -- USE BACKPORT FOR: Python2
from click.types import convert_type
import six
# -----------------------------------------------------------------------------
# PACKAGE META DATA:
# -----------------------------------------------------------------------------
__version__ = "0.2.3"
__author__ = "Jens Engel"
__license__ = "BSD"
# -----------------------------------------------------------------------------
# DECORATORS
# -----------------------------------------------------------------------------
def matches_section(section_name):
"""Decorator for SectionSchema classes to define the mapping between
a config section schema class and one or more config sections with
matching name(s).
.. sourcecode::
@matches_section("foo")
class FooSchema(SectionSchema):
pass
@matches_section(["bar", "baz.*"])
class BarAndBazSchema(SectionSchema):
pass
.. sourcecode:: ini
# -- FILE: *.ini
[foo] # USE: FooSchema
...
[bar] # USE: BarAndBazSchema
...
[baz.alice] # USE: BarAndBazSchema
...
"""
section_names = section_name
if isinstance(section_name, six.string_types):
section_names = [section_name]
elif not isinstance(section_name, (list, tuple)):
raise ValueError("%r (expected: string, strings)" % section_name)
def decorator(cls):
class_section_names = getattr(cls, "section_names", None)
if class_section_names is None:
cls.section_names = list(section_names)
else:
# -- BETTER SUPPORT: For multiple decorators
# @matches_section("foo")
# @matches_section("bar.*")
# class Example(SectionSchema):
# pass
# assert Example.section_names == ["foo", "bar.*"]
approved = [name for name in section_names
if name not in cls.section_names]
cls.section_names = approved + cls.section_names
return cls
return decorator
def assign_param_names(cls=None, param_class=None):
"""Class decorator to assign parameter name to instances of :class:`Param`.
.. sourcecode::
@assign_param_names
class ConfigSectionSchema(object):
alice = Param(type=str)
bob = Param(type=str)
assert ConfigSectionSchema.alice.name == "alice"
assert ConfigSectionSchema.bob.name == "bob"
.. sourcecode::
# -- NESTED ASSIGN: Covers also nested SectionSchema subclasses.
@assign_param_names
class ConfigSectionSchema(object):
class Foo(SectionSchema):
alice = Param(type=str)
bob = Param(type=str)
assert ConfigSectionSchema.Foo.alice.name == "alice"
assert ConfigSectionSchema.Foo.bob.name == "bob"
"""
if param_class is None:
param_class = Param
def decorate_class(cls):
for name, value in select_params_from_section_schema(cls, param_class,
deep=True):
# -- ANNOTATE PARAM: By assigning its name
if not value.name:
value.name = name
return cls
# -- DECORATOR LOGIC:
if cls is None:
# -- CASE: @assign_param_names
# -- CASE: @assign_param_names(...)
return decorate_class
else:
# -- CASE: @assign_param_names class X: ...
# -- CASE: assign_param_names(my_class)
# -- CASE: my_class = assign_param_names(my_class)
return decorate_class(cls)
# -----------------------------------------------------------------------------
# CONFIG SECTION SCHEMA DESCRIPTION
# -----------------------------------------------------------------------------
class SectionSchema(object):
"""Marker for configuration section schema classes."""
# -- ONLY IN DERIVED CLASSES:
# section_names = [] # Section names (or patterns) to apply SectionSchema.
@classmethod
def matches_section(cls, section_name, supported_section_names=None):
"""Indicates if this schema can be used for a config section
by using the section name.
:param section_name: Config section name to check.
:return: True, if this schema can be applied to the config section.
:return: Fals, if this schema does not match the config section.
"""
if supported_section_names is None:
supported_section_names = getattr(cls, "section_names", None)
# pylint: disable=invalid-name
for supported_section_name_or_pattern in supported_section_names:
if fnmatch(section_name, supported_section_name_or_pattern):
return True
# -- OTHERWISE:
return False
class Param(object):
"""Simple Parameter class used for description config file schema params.
Simple replacement for :class:`click.Parameter` or its variants.
Click types are supported.
.. sourcecode::
class FooSchema(SectionSchema):
name = Param(type=str)
flag = Param(type=bool, default=True)
numbers = Param(type=int, multiple=True)
filenames = Param(type=click.Path(), multiple=True)
.. sourcecode:: ini
# -- FILE: foo.ini
[foo]
flag = yes # -- SUPPORTS: true, false, yes, no (case-insensitive)
name = Alice and Bob
numbers = 1 4 9 16 25
filenames = foo/xxx.txt
bar/baz/zzz.txt
"""
# pylint: disable=redefined-builtin
def __init__(self, name=None, type=None, multiple=None, default=None,
help=None):
self.name = name
self.type = convert_type(type, default)
self.multiple = multiple
self.default = default
self.help = help
def parse(self, text):
if self.multiple:
parts = text.strip().split()
values = [self.type.convert(value, self, ctx=None)
for value in parts]
return values
else:
return self.type.convert(text, self, ctx=None)
# -----------------------------------------------------------------------------
# PARSING CONFIG SECTIONS WITH SCHEMA DESCRIPTION
# -----------------------------------------------------------------------------
def select_params_from_section_schema(section_schema, param_class=Param,
deep=False):
"""Selects the parameters of a config section schema.
:param section_schema: Configuration file section schema to use.
:return: Generator of params
"""
# pylint: disable=invalid-name
for name, value in inspect.getmembers(section_schema):
if name.startswith("__") or value is None:
continue # pragma: no cover
elif inspect.isclass(value) and deep:
# -- CASE: class => SELF-CALL (recursively).
# pylint: disable= bad-continuation
cls = value
for name, value in select_params_from_section_schema(cls,
param_class=param_class, deep=True):
yield (name, value)
elif isinstance(value, param_class):
yield (name, value)
def parse_config_section(config_section, section_schema):
"""Parse a config file section (INI file) by using its schema/description.
.. sourcecode::
import configparser # -- NOTE: Use backport for Python2
import click
from click_configfile import SectionSchema, Param, parse_config_section
class ConfigSectionSchema(object):
class Foo(SectionSchema):
name = Param(type=str)
flag = Param(type=bool)
numbers = Param(type=int, multiple=True)
filenames = Param(type=click.Path(), multiple=True)
parser = configparser.ConfigParser()
parser.read(["foo.ini"])
config_section = parser["foo"]
data = parse_config_section(config_section, ConfigSectionSchema.Foo)
# -- FAILS WITH: click.BadParameter if conversion errors occur.
.. sourcecode:: ini
# -- FILE: foo.ini
[foo]
name = Alice
flag = yes # true, false, yes, no (case-insensitive)
numbers = 1 4 9 16 25
filenames = foo/xxx.txt
bar/baz/zzz.txt
:param config_section: Config section to parse
:param section_schema: Schema/description of config section (w/ Param).
:return: Retrieved data, values converted to described types.
:raises: click.BadParameter, if conversion error occurs.
"""
storage = {}
for name, param in select_params_from_section_schema(section_schema):
value = config_section.get(name, None)
if value is None:
if param.default is None:
continue
value = param.default
else:
value = param.parse(value)
# -- DIAGNOSTICS:
# print(" %s = %s" % (name, repr(value)))
storage[name] = value
return storage
# -----------------------------------------------------------------------------
# SUPPORT: READ CONFIGFILE
# -----------------------------------------------------------------------------
def generate_configfile_names(config_files, config_searchpath=None):
"""Generates all configuration file name combinations to read.
.. sourcecode::
# -- ALGORITHM:
# First basenames/directories are prefered and override other files.
for config_path in reversed(config_searchpath):
for config_basename in reversed(config_files):
config_fname = os.path.join(config_path, config_basename)
if os.path.isfile(config_fname):
yield config_fname
:param config_files: List of config file basenames.
:param config_searchpath: List of directories to look for config files.
:return: List of available configuration file names (as generator)
"""
if config_searchpath is None:
config_searchpath = ["."]
for config_path in reversed(config_searchpath):
for config_basename in reversed(config_files):
config_fname = os.path.join(config_path, config_basename)
if os.path.isfile(config_fname):
# MAYBE: yield os.path.normpath(config_fname)
yield config_fname
def select_config_sections(configfile_sections, desired_section_patterns):
"""Select a subset of the sections in a configuration file by using
a list of section names of list of section name patters
(supporting :mod:`fnmatch` wildcards).
:param configfile_sections: List of config section names (as strings).
:param desired_section_patterns:
:return: List of selected section names or empty list (as generator).
"""
for section_name in configfile_sections:
for desired_section_pattern in desired_section_patterns:
if fnmatch(section_name, desired_section_pattern):
yield section_name
# -----------------------------------------------------------------------------
# BOILER-PLATE FOR CONFIG-FILE READER
# -----------------------------------------------------------------------------
class ConfigFileReader(object):
"""Generic configuration file reader.
Concrete configuration file reader class must extend it and specify the
concrete details.
EXAMPLE:
.. sourcecode::
# -- FILE: hello_command.py
from click_configfile import ConfigFileReader, Param, SectionSchema
from click_configfile import matches_section
class ConfigSectionSchema(object):
@matches_section("hello")
class Hello(SectionSchema):
name = Param(type=str)
@matches_section("hello.more.*")
class HelloMore(SectionSchema):
numbers = Param(type=int, multiple=True)
class ConfigFileProcessor(ConfigFileReader):
config_files = ["hello.ini", "hello.cfg"]
config_section_schemas = [
ConfigSectionSchema.Hello, # PRIMARY SCHEMA
ConfigSectionSchema.HelloMore,
]
@classmethod
def get_storage_name_for(cls, section_name):
if section_name == "hello":
return "" # -- MERGE-INTO-STORAGE
elif section_name.startswith("hello.more."):
# -- EXAMPLE: hello.more.alice -> alice
return section_name.replace("hello.more.", "", 1)
# -- OTHERWISE: Delegate to BaseClass
return ConfigFileReader.get_storage_name_for(section_name)
# OR: raise LookupError(section_name)
# -- COMMANDS:
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", type=str)
@click.pass_context
def hello(ctx, name):
pass
"""
config_files = [] # Config filename variants (basenames).
config_section_schemas = [] # Config section schema description.
config_sections = [] # OPTIONAL: Config sections of interest.
config_searchpath = ["."] # OPTIONAL: Where to look for config files.
# -- GENERIC PART:
# Uses declarative specification from above (config_files, config_sections, ...)
@classmethod
def read_config(cls):
configfile_names = list(
generate_configfile_names(cls.config_files, cls.config_searchpath))
parser = configparser.ConfigParser()
parser.optionxform = str
parser.read(configfile_names)
if not cls.config_sections:
# -- AUTO-DISCOVER (once): From cls.config_section_schemas
cls.config_sections = cls.collect_config_sections_from_schemas()
storage = {}
for section_name in select_config_sections(parser.sections(),
cls.config_sections):
# print("PROCESS-SECTION: %s" % section_name)
config_section = parser[section_name]
cls.process_config_section(config_section, storage)
return storage
@classmethod
def collect_config_sections_from_schemas(cls, config_section_schemas=None):
# pylint: disable=invalid-name
"""Derive support config section names from config section schemas.
If no :param:`config_section_schemas` are provided, the schemas from
this class are used (normally defined in the DerivedClass).
:param config_section_schemas: List of config section schema classes.
:return: List of config section names or name patterns (as string).
"""
if config_section_schemas is None:
config_section_schemas = cls.config_section_schemas
collected = []
for schema in config_section_schemas:
collected.extend(schema.section_names)
# -- MAYBE BETTER:
# for name in schema.section_names:
# if name not in collected:
# collected.append(name)
return collected
# -- SPECIFIC PART:
# Specifies which schema to use and where data should be stored.
@classmethod
def process_config_section(cls, config_section, storage):
"""Process the config section and store the extracted data in
the param:`storage` (as outgoing param).
"""
# -- CONCEPT:
# if not storage:
# # -- INIT DATA: With default parts.
# storage.update(dict(_PERSONS={}))
schema = cls.select_config_schema_for(config_section.name)
if not schema:
message = "No schema found for: section=%s"
raise LookupError(message % config_section.name)
# -- PARSE AND STORE CONFIG SECTION:
section_storage = cls.select_storage_for(config_section.name, storage)
section_data = parse_config_section(config_section, schema)
section_storage.update(section_data)
@classmethod
def select_config_schema_for(cls, section_name):
"""Select the config schema that matches the config section (by name).
:param section_name: Config section name (as key).
:return: Config section schmema to use (subclass of: SectionSchema).
"""
# pylint: disable=cell-var-from-loop, redefined-outer-name
for section_schema in cls.config_section_schemas:
schema_matches = getattr(section_schema, "matches_section", None)
if schema_matches is None:
# -- OTHER SCHEMA CLASS: Reuse SectionSchema functionality.
schema_matches = lambda name: SectionSchema.matches_section(
name, section_schema.section_names)
if schema_matches(section_name):
return section_schema
return None
@classmethod
def get_storage_name_for(cls, section_name):
"""Selects where to store the retrieved data from the config section.
:param section_name: Config section name to process (as string).
:return: EMPTY-STRING or None, indicates MERGE-WITH-DEFAULTS.
:return: NON-EMPTY-STRING, for key in default_map to use.
"""
if cls.config_sections and cls.config_sections[0] == section_name:
# -- PRIMARY-SECTION: Merge into storage (defaults_map).
return ""
else:
return section_name
# -- MAYBE BETTER:
# raise LookupError("Unknown section: %s" % section_name)
@classmethod
def select_storage_for(cls, section_name, storage):
"""Selects the data storage for a config section within the
:param:`storage`. The primary config section is normally merged into
the :param:`storage`.
:param section_name: Config section (name) to process.
:param storage: Data storage to use.
:return: :param:`storage` or a part of it (as section storage).
"""
section_storage = storage
storage_name = cls.get_storage_name_for(section_name)
if storage_name:
section_storage = storage.get(storage_name, None)
if section_storage is None:
section_storage = storage[storage_name] = dict()
return section_storage