-
Notifications
You must be signed in to change notification settings - Fork 23
/
config.py
332 lines (297 loc) · 12.7 KB
/
config.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
import re
from logging.config import BaseConfigurator
import picologging
from picologging.handlers import MemoryHandler
IDENTIFIER = re.compile("^[a-z_][a-z0-9_]*$", re.I)
def valid_ident(s):
m = IDENTIFIER.match(s)
if not m:
raise ValueError("Not a valid Python identifier: %r" % s)
return True
def _resolve(name):
"""Resolve a dotted name to a global object."""
name = name.split(".")
used = name.pop(0)
found = __import__(used)
for n in name:
used = used + "." + n
try:
found = getattr(found, n)
except AttributeError:
__import__(used)
found = getattr(found, n)
return found
def _handle_existing_loggers(existing, child_loggers, disable_existing):
"""
When (re)configuring logging, handle loggers which were in the previous
configuration but are not in the new configuration. There's no point
deleting them as other threads may continue to hold references to them;
and by disabling them, you stop them doing any logging.
However, don't disable children of named loggers, as that's probably not
what was intended by the user. Also, allow existing loggers to NOT be
disabled if disable_existing is false.
"""
root = picologging.root
for log in existing:
logger = root.manager.loggerDict[log]
if log in child_loggers:
if not isinstance(logger, picologging.PlaceHolder):
logger.setLevel(picologging.NOTSET)
logger.handlers = []
logger.propagate = True
else:
logger.disabled = disable_existing
class DictConfigurator(BaseConfigurator):
"""
Configure logging using a dictionary-like object to describe the
configuration.
"""
def configure(self):
config = self.config
if "version" not in config:
raise ValueError("dictionary doesn't specify a version")
if config["version"] != 1:
raise ValueError("Unsupported version: %s" % config["version"])
incremental = config.pop("incremental", False)
if incremental:
raise ValueError("Incremental option is not supported.")
EMPTY_DICT = {}
disable_existing = config.pop("disable_existing_loggers", True)
# Do formatters first - they don't refer to anything else
formatters = config.get("formatters", EMPTY_DICT)
for name in formatters:
try:
formatters[name] = self.configure_formatter(formatters[name])
except Exception as e:
raise ValueError("Unable to configure " "formatter %r" % name) from e
# Next, do filters - they don't refer to anything else, either
filters = config.get("filters", EMPTY_DICT)
for name in filters:
try:
filters[name] = self.configure_filter(filters[name])
except Exception as e:
raise ValueError("Unable to configure " "filter %r" % name) from e
# Next, do handlers - they refer to formatters and filters
# As handlers can refer to other handlers, sort the keys
# to allow a deterministic order of configuration
handlers = config.get("handlers", EMPTY_DICT)
deferred = []
for name in sorted(handlers):
try:
handler = self.configure_handler(handlers[name])
handler.name = name
handlers[name] = handler
except Exception as e:
if "target not configured yet" in str(e.__cause__):
deferred.append(name)
else:
raise ValueError("Unable to configure handler " "%r" % name) from e
# Now do any that were deferred
for name in deferred:
try:
handler = self.configure_handler(handlers[name])
handler.name = name
handlers[name] = handler
except Exception as e:
raise ValueError("Unable to configure handler " "%r" % name) from e
# Next, do loggers - they refer to handlers and filters
# we don't want to lose the existing loggers,
# since other threads may have pointers to them.
# existing is set to contain all existing loggers,
# and as we go through the new configuration we
# remove any which are configured. At the end,
# what's left in existing is the set of loggers
# which were in the previous configuration but
# which are not in the new configuration.
root = picologging.root
existing = list(root.manager.loggerDict.keys())
# The list needs to be sorted so that we can
# avoid disabling child loggers of explicitly
# named loggers. With a sorted list it is easier
# to find the child loggers.
existing.sort()
# We'll keep the list of existing loggers
# which are children of named loggers here...
child_loggers = []
# now set up the new ones...
loggers = config.get("loggers", EMPTY_DICT)
for name in loggers:
if name in existing:
i = existing.index(name) + 1 # look after name
prefixed = name + "."
pflen = len(prefixed)
num_existing = len(existing)
while i < num_existing:
if existing[i][:pflen] == prefixed:
child_loggers.append(existing[i])
i += 1
existing.remove(name)
try:
self.configure_logger(name, loggers[name])
except Exception as e:
raise ValueError("Unable to configure logger " "%r" % name) from e
# Disable any old loggers. There's no point deleting
# them as other threads may continue to hold references
# and by disabling them, you stop them doing any logging.
# However, don't disable children of named loggers, as that's
# probably not what was intended by the user.
_handle_existing_loggers(existing, child_loggers, disable_existing)
# And finally, do the root logger
root = config.get("root", None)
if root:
try:
self.configure_root(root)
except Exception as e:
raise ValueError("Unable to configure root " "logger") from e
def configure_formatter(self, config):
"""Configure a formatter from a dictionary."""
if "()" in config:
factory = config["()"] # for use in exception handler
try:
result = self.configure_custom(config)
except TypeError as te:
if "'format'" not in str(te):
raise
# Name of parameter changed from fmt to format.
# Retry with old name.
# This is so that code can be used with older Python versions
# (e.g. by Django)
config["fmt"] = config.pop("format")
config["()"] = factory
result = self.configure_custom(config)
else:
fmt = config.get("format", None)
dfmt = config.get("datefmt", None)
style = config.get("style", "%")
cname = config.get("class", None)
if not cname:
c = picologging.Formatter
else:
c = _resolve(cname)
# A TypeError would be raised if "validate" key is passed in with a formatter callable
# that does not accept "validate" as a parameter
if (
"validate" in config
): # if user hasn't mentioned it, the default will be fine
result = c(fmt, dfmt, style, config["validate"])
else:
result = c(fmt, dfmt, style)
return result
def configure_filter(self, config):
"""Configure a filter from a dictionary."""
if "()" in config:
result = self.configure_custom(config)
else:
name = config.get("name", "")
result = picologging.Filter(name)
return result
def add_filters(self, filterer, filters):
"""Add filters to a filterer from a list of names."""
for f in filters:
try:
filterer.addFilter(self.config["filters"][f])
except Exception as e:
raise ValueError("Unable to add filter %r" % f) from e
def configure_handler(self, config):
"""Configure a handler from a dictionary."""
config_copy = dict(config) # for restoring in case of error
formatter = config.pop("formatter", None)
if formatter:
try:
formatter = self.config["formatters"][formatter]
except Exception as e:
raise ValueError("Unable to set formatter " "%r" % formatter) from e
level = config.pop("level", None)
filters = config.pop("filters", None)
if "()" in config:
c = config.pop("()")
if not callable(c):
c = self.resolve(c)
factory = c
else:
cname = config.pop("class")
klass = self.resolve(cname)
# Special case for handler which refers to another handler
if issubclass(klass, MemoryHandler) and "target" in config:
try:
th = self.config["handlers"][config["target"]]
if not isinstance(th, picologging.Handler):
config.update(config_copy) # restore for deferred cfg
raise TypeError("target not configured yet")
config["target"] = th
except Exception as e:
raise ValueError(
"Unable to set target handler " "%r" % config["target"]
) from e
# elif (
# issubclass(klass, picologging.handlers.SMTPHandler)
# and "mailhost" in config
# ):
# config["mailhost"] = self.as_tuple(config["mailhost"])
# elif (
# issubclass(klass, picologging.handlers.SysLogHandler)
# and "address" in config
# ):
# config["address"] = self.as_tuple(config["address"])
factory = klass
props = config.pop(".", None)
kwargs = {k: config[k] for k in config if valid_ident(k)}
try:
result = factory(**kwargs)
except TypeError as te:
if "'stream'" not in str(te):
raise
# The argument name changed from strm to stream
# Retry with old name.
# This is so that code can be used with older Python versions
# (e.g. by Django)
kwargs["strm"] = kwargs.pop("stream")
result = factory(**kwargs)
if formatter:
result.setFormatter(formatter)
if level is not None:
result.setLevel(picologging._checkLevel(level))
if filters:
self.add_filters(result, filters)
if props:
for name, value in props.items():
setattr(result, name, value)
return result
def add_handlers(self, logger, handlers):
"""Add handlers to a logger from a list of names."""
for h in handlers:
try:
logger.addHandler(self.config["handlers"][h])
except Exception as e:
raise ValueError("Unable to add handler %r" % h) from e
def common_logger_config(self, logger, config):
"""
Perform configuration which is common to root and non-root loggers.
"""
level = config.get("level", None)
if level is not None:
logger.setLevel(picologging._checkLevel(level))
# Remove any existing handlers
for h in logger.handlers[:]:
logger.removeHandler(h)
handlers = config.get("handlers", None)
if handlers:
self.add_handlers(logger, handlers)
filters = config.get("filters", None)
if filters:
self.add_filters(logger, filters)
def configure_logger(self, name, config):
"""Configure a non-root logger from a dictionary."""
logger = picologging.getLogger(name)
self.common_logger_config(logger, config)
propagate = config.get("propagate", None)
if propagate is not None:
logger.propagate = propagate
def configure_root(self, config):
"""Configure a root logger from a dictionary."""
root = picologging.getLogger()
self.common_logger_config(root, config)
dictConfigClass = DictConfigurator
def dictConfig(config):
"""Configure logging using a dictionary."""
dictConfigClass(config).configure()