-
-
Notifications
You must be signed in to change notification settings - Fork 2.4k
/
base.py
381 lines (295 loc) · 12.4 KB
/
base.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
from __future__ import annotations
import functools
import logging
import os
import sys
from collections import UserDict
from contextlib import contextmanager
from typing import (
IO,
TYPE_CHECKING,
Generic,
Iterator,
List,
Optional,
Sequence,
Tuple,
TypeVar,
Union,
overload,
)
from yaml import YAMLError
from mkdocs import exceptions, utils
if TYPE_CHECKING:
from mkdocs.config.defaults import MkDocsConfig
log = logging.getLogger('mkdocs.config')
T = TypeVar('T')
class BaseConfigOption(Generic[T]):
def __init__(self):
self.warnings: List[str] = []
self.default = None
@property
def default(self):
try:
# ensure no mutable values are assigned
return self._default.copy()
except AttributeError:
return self._default
@default.setter
def default(self, value):
self._default = value
def validate(self, value: object) -> T:
return self.run_validation(value)
def reset_warnings(self) -> None:
self.warnings = []
def pre_validation(self, config: Config, key_name: str) -> None:
"""
Before all options are validated, perform a pre-validation process.
The pre-validation process method should be implemented by subclasses.
"""
def run_validation(self, value: object):
"""
Perform validation for a value.
The run_validation method should be implemented by subclasses.
"""
return value
def post_validation(self, config: Config, key_name: str) -> None:
"""
After all options have passed validation, perform a post-validation
process to do any additional changes dependent on other config values.
The post-validation process method should be implemented by subclasses.
"""
def __set_name__(self, owner, name):
self._name = name
@overload
def __get__(self, obj: Config, type=None) -> T:
...
@overload
def __get__(self, obj, type=None) -> BaseConfigOption:
...
def __get__(self, obj, type=None):
if not isinstance(obj, Config):
return self
return obj[self._name]
def __set__(self, obj, value: T):
if not isinstance(obj, Config):
raise AttributeError(
f"can't set attribute ({self._name}) because the parent is a {type(obj)} not a {Config}"
)
obj[self._name] = value
class ValidationError(Exception):
"""Raised during the validation process of the config on errors."""
def __eq__(self, other):
return type(self) is type(other) and str(self) == str(other)
PlainConfigSchemaItem = Tuple[str, BaseConfigOption]
PlainConfigSchema = Sequence[PlainConfigSchemaItem]
ConfigErrors = List[Tuple[str, Exception]]
ConfigWarnings = List[Tuple[str, str]]
class Config(UserDict):
"""
Base class for MkDocs configuration, plugin configuration (and sub-configuration) objects.
It should be subclassed and have `ConfigOption`s defined as attributes.
For examples, see mkdocs/contrib/search/__init__.py and mkdocs/config/defaults.py.
Behavior as it was prior to MkDocs 1.4 is now handled by LegacyConfig.
"""
_schema: PlainConfigSchema
config_file_path: Optional[str]
def __init_subclass__(cls):
schema = dict(getattr(cls, '_schema', ()))
for attr_name, attr in cls.__dict__.items():
if isinstance(attr, BaseConfigOption):
schema[attr_name] = attr
cls._schema = tuple(schema.items())
for attr_name, attr in cls._schema:
attr.required = True
if getattr(attr, '_legacy_required', None) is not None:
raise TypeError(
f"{cls.__name__}.{attr_name}: "
"Setting 'required' is unsupported in class-based configs. "
"All values are required, or can be wrapped into config_options.Optional"
)
def __new__(cls, *args, **kwargs) -> Config:
"""Compatibility: allow referring to `LegacyConfig(...)` constructor as `Config(...)`."""
if cls is Config:
return LegacyConfig(*args, **kwargs)
return super().__new__(cls)
def __init__(self, config_file_path: Optional[Union[str, bytes]] = None):
super().__init__()
self.user_configs: List[dict] = []
self.set_defaults()
self._schema_keys = {k for k, v in self._schema}
# Ensure config_file_path is a Unicode string
if config_file_path is not None and not isinstance(config_file_path, str):
try:
# Assume config_file_path is encoded with the file system encoding.
config_file_path = config_file_path.decode(encoding=sys.getfilesystemencoding())
except UnicodeDecodeError:
raise ValidationError("config_file_path is not a Unicode string.")
self.config_file_path = config_file_path
def set_defaults(self) -> None:
"""
Set the base config by going through each validator and getting the
default if it has one.
"""
for key, config_option in self._schema:
self[key] = config_option.default
def _validate(self) -> Tuple[ConfigErrors, ConfigWarnings]:
failed: ConfigErrors = []
warnings: ConfigWarnings = []
for key, config_option in self._schema:
try:
value = self.get(key)
self[key] = config_option.validate(value)
warnings.extend((key, w) for w in config_option.warnings)
config_option.reset_warnings()
except ValidationError as e:
failed.append((key, e))
for key in set(self.keys()) - self._schema_keys:
warnings.append((key, f"Unrecognised configuration name: {key}"))
return failed, warnings
def _pre_validate(self) -> Tuple[ConfigErrors, ConfigWarnings]:
failed: ConfigErrors = []
warnings: ConfigWarnings = []
for key, config_option in self._schema:
try:
config_option.pre_validation(self, key_name=key)
warnings.extend((key, w) for w in config_option.warnings)
config_option.reset_warnings()
except ValidationError as e:
failed.append((key, e))
return failed, warnings
def _post_validate(self) -> Tuple[ConfigErrors, ConfigWarnings]:
failed: ConfigErrors = []
warnings: ConfigWarnings = []
for key, config_option in self._schema:
try:
config_option.post_validation(self, key_name=key)
warnings.extend((key, w) for w in config_option.warnings)
config_option.reset_warnings()
except ValidationError as e:
failed.append((key, e))
return failed, warnings
def validate(self) -> Tuple[ConfigErrors, ConfigWarnings]:
failed, warnings = self._pre_validate()
run_failed, run_warnings = self._validate()
failed.extend(run_failed)
warnings.extend(run_warnings)
# Only run the post validation steps if there are no failures, warnings
# are okay.
if len(failed) == 0:
post_failed, post_warnings = self._post_validate()
failed.extend(post_failed)
warnings.extend(post_warnings)
return failed, warnings
def load_dict(self, patch: Optional[dict]) -> None:
"""Load config options from a dictionary."""
if not isinstance(patch, dict):
raise exceptions.ConfigurationError(
"The configuration is invalid. The expected type was a key "
"value mapping (a python dict) but we got an object of type: "
f"{type(patch)}"
)
self.user_configs.append(patch)
self.update(patch)
def load_file(self, config_file: IO) -> None:
"""Load config options from the open file descriptor of a YAML file."""
try:
return self.load_dict(utils.yaml_load(config_file))
except YAMLError as e:
# MkDocs knows and understands ConfigurationErrors
raise exceptions.ConfigurationError(
f"MkDocs encountered an error parsing the configuration file: {e}"
)
@functools.lru_cache(maxsize=None)
def get_schema(cls: type) -> PlainConfigSchema:
"""
Extract ConfigOptions defined in a class (used just as a container) and put them into a schema tuple.
"""
if issubclass(cls, Config):
return cls._schema
return tuple((k, v) for k, v in cls.__dict__.items() if isinstance(v, BaseConfigOption))
class LegacyConfig(Config):
"""
A configuration object for plugins, as just a dict without type-safe attribute access.
"""
def __init__(self, schema: PlainConfigSchema, config_file_path: Optional[str] = None):
self._schema = tuple((k, v) for k, v in schema) # Re-create just for validation
super().__init__(config_file_path)
@contextmanager
def _open_config_file(config_file: Optional[Union[str, IO]]) -> Iterator[IO]:
"""
A context manager which yields an open file descriptor ready to be read.
Accepts a filename as a string, an open or closed file descriptor, or None.
When None, it defaults to `mkdocs.yml` in the CWD. If a closed file descriptor
is received, a new file descriptor is opened for the same file.
The file descriptor is automatically closed when the context manager block is existed.
"""
# Default to the standard config filename.
if config_file is None:
paths_to_try = ['mkdocs.yml', 'mkdocs.yaml']
# If it is a string, we can assume it is a path and attempt to open it.
elif isinstance(config_file, str):
paths_to_try = [config_file]
# If closed file descriptor, get file path to reopen later.
elif getattr(config_file, 'closed', False):
paths_to_try = [config_file.name]
else:
result_config_file = config_file
paths_to_try = None
if paths_to_try:
# config_file is not a file descriptor, so open it as a path.
for path in paths_to_try:
path = os.path.abspath(path)
log.debug(f"Loading configuration file: {path}")
try:
result_config_file = open(path, 'rb')
break
except FileNotFoundError:
continue
else:
raise exceptions.ConfigurationError(f"Config file '{paths_to_try[0]}' does not exist.")
else:
log.debug(f"Loading configuration file: {result_config_file}")
# Ensure file descriptor is at beginning
result_config_file.seek(0)
try:
yield result_config_file
finally:
if hasattr(result_config_file, 'close'):
result_config_file.close()
def load_config(config_file: Optional[Union[str, IO]] = None, **kwargs) -> MkDocsConfig:
"""
Load the configuration for a given file object or name
The config_file can either be a file object, string or None. If it is None
the default `mkdocs.yml` filename will loaded.
Extra kwargs are passed to the configuration to replace any default values
unless they themselves are None.
"""
options = kwargs.copy()
# Filter None values from the options. This usually happens with optional
# parameters from Click.
for key, value in options.copy().items():
if value is None:
options.pop(key)
with _open_config_file(config_file) as fd:
# Initialize the config with the default schema.
from mkdocs.config.defaults import MkDocsConfig
cfg = MkDocsConfig(config_file_path=getattr(fd, 'name', ''))
# load the config file
cfg.load_file(fd)
# Then load the options to overwrite anything in the config.
cfg.load_dict(options)
errors, warnings = cfg.validate()
for config_name, warning in warnings:
log.warning(f"Config value: '{config_name}'. Warning: {warning}")
for config_name, error in errors:
log.error(f"Config value: '{config_name}'. Error: {error}")
for key, value in cfg.items():
log.debug(f"Config value: '{key}' = {value!r}")
if len(errors) > 0:
raise exceptions.Abort(f"Aborted with {len(errors)} Configuration Errors!")
elif cfg['strict'] and len(warnings) > 0:
raise exceptions.Abort(
f"Aborted with {len(warnings)} Configuration Warnings in 'strict' mode!"
)
return cfg