# Copyright 2001-2019 by Vinay Sajip. All Rights Reserved.
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose and without fee is hereby granted,
# provided that the above copyright notice appear in all copies and that
# both that copyright notice and this permission notice appear in
# supporting documentation, and that the name of Vinay Sajip
# not be used in advertising or publicity pertaining to distribution
# of the software without specific, written prior permission.
# VINAY SAJIP DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, INCLUDING
# ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# VINAY SAJIP BE LIABLE FOR ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR
# ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER
# IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
Configuration functions for the logging package for Python. The core package
is based on PEP 282 and comments thereto in comp.lang.python, and influenced
by Apache's log4j system.
Copyright (C) 2001-2019 Vinay Sajip. All Rights Reserved.
To use, simply 'import logging' and log away!
from socketserver import ThreadingTCPServer, StreamRequestHandler
DEFAULT_LOGGING_CONFIG_PORT = 9030
RESET_ERROR = errno.ECONNRESET
# The following code implements a socket listener for on-the-fly
# reconfiguration of logging.
# _listener holds the server object doing the listening
def fileConfig(fname, defaults=None, disable_existing_loggers=True):
Read the logging configuration from a ConfigParser-format file.
This can be called several times from an application, allowing an end user
the ability to select from various pre-canned configurations (if the
developer provides a mechanism to present the choices and load the chosen
if isinstance(fname, configparser.RawConfigParser):
cp = configparser.ConfigParser(defaults)
if hasattr(fname, 'readline'):
formatters = _create_formatters(cp)
# Handlers add themselves to logging._handlers
handlers = _install_handlers(cp, formatters)
_install_loggers(cp, handlers, disable_existing_loggers)
"""Resolve a dotted name to a global object."""
found = getattr(found, n)
found = getattr(found, n)
def _strip_spaces(alist):
return map(str.strip, alist)
def _create_formatters(cp):
"""Create and return formatters"""
flist = cp["formatters"]["keys"]
flist = _strip_spaces(flist)
sectname = "formatter_%s" % form
fs = cp.get(sectname, "format", raw=True, fallback=None)
dfs = cp.get(sectname, "datefmt", raw=True, fallback=None)
stl = cp.get(sectname, "style", raw=True, fallback='%')
class_name = cp[sectname].get("class")
def _install_handlers(cp, formatters):
"""Install and return handlers"""
hlist = cp["handlers"]["keys"]
hlist = _strip_spaces(hlist)
fixups = [] #for inter-handler references
section = cp["handler_%s" % hand]
fmt = section.get("formatter", "")
klass = eval(klass, vars(logging))
except (AttributeError, NameError):
args = section.get("args", '()')
args = eval(args, vars(logging))
kwargs = section.get("kwargs", '{}')
kwargs = eval(kwargs, vars(logging))
h = klass(*args, **kwargs)
h.setFormatter(formatters[fmt])
if issubclass(klass, logging.handlers.MemoryHandler):
target = section.get("target", "")
if len(target): #the target handler may not be loaded yet, so keep for later...
fixups.append((h, target))
#now all handlers are loaded, fixup inter-handler references...
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.
logger = root.manager.loggerDict[log]
if not isinstance(logger, logging.PlaceHolder):
logger.setLevel(logging.NOTSET)
logger.disabled = disable_existing
def _install_loggers(cp, handlers, disable_existing):
"""Create and install loggers"""
# configure the root first
llist = cp["loggers"]["keys"]
llist = list(_strip_spaces(llist))
section = cp["logger_root"]
for h in root.handlers[:]:
hlist = section["handlers"]
hlist = _strip_spaces(hlist)
log.addHandler(handlers[hand])
#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.
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.
#We'll keep the list of existing loggers
#which are children of named loggers here...
#now set up the new ones...
section = cp["logger_%s" % log]
propagate = section.getint("propagate", fallback=1)
logger = logging.getLogger(qn)
i = existing.index(qn) + 1 # start with the entry after qn
num_existing = len(existing)
if existing[i][:pflen] == prefixed:
child_loggers.append(existing[i])
for h in logger.handlers[:]:
logger.propagate = propagate
hlist = section["handlers"]
hlist = _strip_spaces(hlist)
logger.addHandler(handlers[hand])
#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.
# logger = root.manager.loggerDict[log]
# if log in child_loggers:
# logger.level = logging.NOTSET
# elif disable_existing_loggers:
_handle_existing_loggers(existing, child_loggers, disable_existing)
def _clearExistingHandlers():
"""Clear and close existing handlers"""
logging._handlers.clear()
logging.shutdown(logging._handlerList[:])
del logging._handlerList[:]
IDENTIFIER = re.compile('^[a-z_][a-z0-9_]*$', re.I)
raise ValueError('Not a valid Python identifier: %r' % s)
class ConvertingMixin(object):
"""For ConvertingXXX's, this mixin class provides common functions"""
def convert_with_key(self, key, value, replace=True):
result = self.configurator.convert(value)
#If the converted value is different, save for next time
if type(result) in (ConvertingDict, ConvertingList,
def convert(self, value):
result = self.configurator.convert(value)
if type(result) in (ConvertingDict, ConvertingList,
# The ConvertingXXX classes are wrappers around standard Python containers,
# and they serve to convert any suitable values in the container. The
# conversion converts base dicts, lists and tuples to their wrapped
# equivalents, whereas strings which match a conversion format are converted
# Each wrapper should have a configurator attribute holding the actual
# configurator to use for conversion.
class ConvertingDict(dict, ConvertingMixin):
"""A converting dictionary wrapper."""
def __getitem__(self, key):
value = dict.__getitem__(self, key)
return self.convert_with_key(key, value)
def get(self, key, default=None):
value = dict.get(self, key, default)
return self.convert_with_key(key, value)
def pop(self, key, default=None):
value = dict.pop(self, key, default)
return self.convert_with_key(key, value, replace=False)
class ConvertingList(list, ConvertingMixin):
"""A converting list wrapper."""
def __getitem__(self, key):
value = list.__getitem__(self, key)
return self.convert_with_key(key, value)
value = list.pop(self, idx)
return self.convert(value)
class ConvertingTuple(tuple, ConvertingMixin):
"""A converting tuple wrapper."""
def __getitem__(self, key):
value = tuple.__getitem__(self, key)
# Can't replace a tuple entry.
return self.convert_with_key(key, value, replace=False)
class BaseConfigurator(object):
The configurator base class which defines some useful defaults.
CONVERT_PATTERN = re.compile(r'^(?P<prefix>[a-z]+)://(?P<suffix>.*)$')
WORD_PATTERN = re.compile(r'^\s*(\w+)\s*')
DOT_PATTERN = re.compile(r'^\.\s*(\w+)\s*')
INDEX_PATTERN = re.compile(r'^\[\s*(\w+)\s*\]\s*')
DIGIT_PATTERN = re.compile(r'^\d+$')
# We might want to use a different one, e.g. importlib
importer = staticmethod(__import__)
def __init__(self, config):
self.config = ConvertingDict(config)
self.config.configurator = self
Resolve strings to objects using standard import and attribute
found = self.importer(used)
found = getattr(found, frag)
found = getattr(found, frag)
e, tb = sys.exc_info()[1:]
v = ValueError('Cannot resolve %r: %s' % (s, e))
v.__cause__, v.__traceback__ = e, tb
def ext_convert(self, value):
"""Default converter for the ext:// protocol."""
return self.resolve(value)
def cfg_convert(self, value):
"""Default converter for the cfg:// protocol."""
m = self.WORD_PATTERN.match(rest)
raise ValueError("Unable to convert %r" % value)
d = self.config[m.groups()[0]]
m = self.DOT_PATTERN.match(rest)
m = self.INDEX_PATTERN.match(rest)
if not self.DIGIT_PATTERN.match(idx):
n = int(idx) # try as number first (most likely)
raise ValueError('Unable to convert '
'%r at %r' % (value, rest))
def convert(self, value):
Convert values to an appropriate type. dicts, lists and tuples are
replaced by their converting alternatives. Strings are checked to
see if they have a conversion format and are converted if they do.
if not isinstance(value, ConvertingDict) and isinstance(value, dict):
value = ConvertingDict(value)
value.configurator = self
elif not isinstance(value, ConvertingList) and isinstance(value, list):
value = ConvertingList(value)
value.configurator = self
elif not isinstance(value, ConvertingTuple) and\
isinstance(value, tuple) and not hasattr(value, '_fields'):
value = ConvertingTuple(value)
value.configurator = self
elif isinstance(value, str): # str for py3k
m = self.CONVERT_PATTERN.match(value)
converter = self.value_converters.get(prefix, None)
converter = getattr(self, converter)
value = converter(suffix)
def configure_custom(self, config):
"""Configure an object with a user-supplied factory."""
props = config.pop('.', None)
# Check for valid identifiers
kwargs = {k: config[k] for k in config if valid_ident(k)}
for name, value in props.items():
setattr(result, name, value)
def as_tuple(self, value):
"""Utility function which converts lists to tuples."""
if isinstance(value, list):
class DictConfigurator(BaseConfigurator):
Configure logging using a dictionary-like object to describe the
"""Do the configuration."""
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)
It is recommended that you Edit text format, this type of Fix handles quite a lot in one request
