import threading
import traceback
+from bisect import bisect_left
from socketserver import ThreadingTCPServer, StreamRequestHandler
what was intended by the user. Also, allow existing loggers to NOT be
disabled if disable_existing is false.
"""
- root = logging.root
for log in existing:
- logger = root.manager.loggerDict[log]
+ logger = logging.root.manager.loggerDict[log]
if log in child_loggers:
if not isinstance(logger, logging.PlaceHolder):
logger.setLevel(logging.NOTSET)
else:
logger.disabled = disable_existing
+def _forget_existing_logger(name, existing, existing_set, child_loggers):
+ """Forget a configured logger and record its existing children."""
+ prefixed = name + "."
+ i = bisect_left(existing, prefixed)
+ num_existing = len(existing)
+ while i < num_existing:
+ child = existing[i]
+ if not child.startswith(prefixed):
+ break
+ if child in existing_set:
+ child_loggers[child] = None
+ i += 1
+ existing_set.remove(name)
+
def _install_loggers(cp, handlers, disable_existing):
"""Create and install loggers"""
#named loggers. With a sorted list it is easier
#to find the child loggers.
existing.sort()
+ existing_set = set(existing)
#We'll keep the list of existing loggers
#which are children of named loggers here...
- child_loggers = []
+ child_loggers = {}
#now set up the new ones...
for log in llist:
section = cp["logger_%s" % log]
qn = section["qualname"]
propagate = section.getint("propagate", fallback=1)
logger = logging.getLogger(qn)
- if qn in existing:
- i = existing.index(qn) + 1 # start with the entry after qn
- prefixed = qn + "."
- 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(qn)
+ if qn in existing_set:
+ _forget_existing_logger(qn, existing, existing_set, child_loggers)
if "level" in section:
level = section["level"]
logger.setLevel(level)
# logger.propagate = 1
# elif disable_existing_loggers:
# logger.disabled = 1
+ existing = [name for name in existing if name in existing_set]
_handle_existing_loggers(existing, child_loggers, disable_existing)
#named loggers. With a sorted list it is easier
#to find the child loggers.
existing.sort()
+ existing_set = set(existing)
#We'll keep the list of existing loggers
#which are children of named loggers here...
- child_loggers = []
+ 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)
+ if name in existing_set:
+ _forget_existing_logger(name, existing, existing_set,
+ child_loggers)
try:
self.configure_logger(name, loggers[name])
except Exception as e:
# logger.propagate = True
# elif disable_existing:
# logger.disabled = True
+ existing = [name for name in existing if name in existing_set]
_handle_existing_loggers(existing, child_loggers,
disable_existing)
# Logger should be enabled, since explicitly mentioned
self.assertFalse(logger.disabled)
+ def test_disable_existing_loggers_preserves_children(self):
+ parent = logging.getLogger('many')
+ child = logging.getLogger('many.child')
+ child.setLevel(logging.CRITICAL)
+ self.assertFalse(child.isEnabledFor(logging.INFO))
+ cousin = logging.getLogger('many-child')
+ for i in range(20):
+ logging.getLogger(f'many-sibling-{i}')
+
+ self.apply_config({
+ 'version': 1,
+ 'loggers': {
+ 'many': {
+ 'level': 'INFO',
+ },
+ },
+ })
+
+ self.assertFalse(parent.disabled)
+ self.assertFalse(child.disabled)
+ self.assertEqual(child.level, logging.NOTSET)
+ self.assertTrue(child.isEnabledFor(logging.INFO))
+ self.assertTrue(cousin.disabled)
+
def test_111615(self):
# See gh-111615
import_helper.import_module('_multiprocessing') # see gh-113692