+class DynamicPerScopeLoggingLevelFilter(logging.Filter):
+ """Only interested in seeing logging messages from an allow list of
+ module names or module:function names. Block others.
+
+ """
+ @staticmethod
+ def level_name_to_level(name: str) -> int:
+ numeric_level = getattr(
+ logging,
+ name,
+ None
+ )
+ if not isinstance(numeric_level, int):
+ raise ValueError('Invalid level: {name}')
+ return numeric_level
+
+ def __init__(
+ self,
+ default_logging_level: int,
+ per_scope_logging_levels: str,
+ ) -> None:
+ super().__init__()
+ self.valid_levels = set(['NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'])
+ self.default_logging_level = default_logging_level
+ self.level_by_scope = {}
+ if per_scope_logging_levels is not None:
+ for chunk in per_scope_logging_levels.split(','):
+ if '=' not in chunk:
+ print(
+ f'Malformed lmodule directive: "{chunk}", missing "=". Ignored.',
+ file=sys.stderr
+ )
+ continue
+ try:
+ (scope, level) = chunk.split('=')
+ except ValueError:
+ print(
+ f'Malformed lmodule directive: "{chunk}". Ignored.',
+ file=sys.stderr
+ )
+ continue
+ scope = scope.strip()
+ level = level.strip().upper()
+ if level not in self.valid_levels:
+ print(
+ f'Malformed lmodule directive: "{chunk}", bad level. Ignored.',
+ file=sys.stderr
+ )
+ continue
+ self.level_by_scope[scope] = (
+ DynamicPerScopeLoggingLevelFilter.level_name_to_level(
+ level
+ )
+ )
+
+ @overrides
+ def filter(self, record: logging.LogRecord) -> bool:
+ # First try to find a logging level by scope (--lmodule)
+ if len(self.level_by_scope) > 0:
+ min_level = None
+ for scope in (
+ record.module,
+ f'{record.module}:{record.funcName}',
+ f':{record.funcName}'
+ ):
+ level = self.level_by_scope.get(scope, None)
+ if level is not None:
+ if min_level is None or level < min_level:
+ min_level = level
+
+ # If we found one, use it instead of the global default level.
+ if min_level is not None:
+ return record.levelno >= min_level
+
+ # Otherwise, use the global logging level (--logging_level)
+ return record.levelno >= self.default_logging_level
+
+