Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def emit(self, level, *args, **kwargs):
if log_levels.index(self._log_level) < log_levels.index(level):
return
if level == "trace":
return self._trace(*args, **kwargs)
level = LogLevel.lookupByName(level)
return self._log(level, *args, **kwargs)
def emit(self, level, *args, **kwargs):
if log_levels.index(self._log_level) < log_levels.index(level):
return
if level == "trace":
return self._trace(*args, **kwargs)
level = LogLevel.lookupByName(level)
return self._log(level, *args, **kwargs)
def _acceptable_level(self, level):
if self._levels is None:
target_level = log_levels.index(_log_level)
self._levels = [
self.to_tx[lvl]
for lvl in log_levels
if log_levels.index(lvl) <= target_level and lvl != "none"
]
return level in self._levels
def _set_level(self, level):
target_level = log_levels.index(level)
# this binds either _log or _no_op above to this instance,
# depending on the desired level.
for (idx, name) in enumerate(log_levels):
if idx < target_level:
log_method = functools.partial(_log, self, name)
else:
log_method = _no_op
setattr(self, name, log_method)
def _acceptable_level(self, level):
if self._levels is None:
target_level = log_levels.index(_log_level)
self._levels = [
self.to_tx[lvl]
for lvl in log_levels
if log_levels.index(lvl) <= target_level and lvl != "none"
]
return level in self._levels
def __init__(self, **kwargs):
self.namespace = kwargs.get('namespace', None)
if _loggers is not None:
_loggers.append(weakref.ref(self))
# this class will get overridden by _TxLogger below, so we
# bind *every* level here; set_log_level will un-bind
# some.
for name in log_levels:
setattr(self, name, partial(self._do_log, name))
def start_logging(out=_stdout, level='info'):
"""
Begin logging.
:param out: if provided, a file-like object to log to. By default, this is
stdout.
:param level: the maximum log-level to emit (a string)
"""
global _log_level, _loggers, _started_logging
if level not in log_levels:
raise RuntimeError(
"Invalid log level '{0}'; valid are: {1}".format(
level, ', '.join(log_levels)
)
)
if _started_logging:
return
_started_logging = True
_log_level = level
handler = _TxaioFileHandler(out)
logging.getLogger().addHandler(handler)
# note: Don't need to call basicConfig() or similar, because we've
# now added at least one handler to the root logger
logging.raiseExceptions = True # FIXME
level_to_stdlib = {
'critical': logging.CRITICAL,
def _set_log_level(self, level):
target_level = log_levels.index(level)
# this binds either _log or _no_op above to this instance,
# depending on the desired level.
for (idx, name) in enumerate(log_levels):
if idx <= target_level:
log_method = functools.partial(_log, self, name)
else:
log_method = _no_op
setattr(self, name, log_method)
self._log_level = level
def _acceptable_level(self, level):
if self._levels is None:
target_level = log_levels.index(_log_level)
self._levels = [
self.to_tx[lvl]
for lvl in log_levels
if log_levels.index(lvl) <= target_level and lvl != "none"
]
return level in self._levels