/usr/share/pyshared/kombu/log.py is in python-kombu 2.1.8-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | from __future__ import absolute_import
import os
import logging
import sys
from .utils import cached_property
from .utils.compat import WatchedFileHandler
from .utils.encoding import safe_repr, safe_str
from .utils.functional import maybe_promise
__all__ = ["LogMixin", "LOG_LEVELS", "get_loglevel", "setup_logging"]
LOG_LEVELS = dict(logging._levelNames)
LOG_LEVELS["FATAL"] = logging.FATAL
LOG_LEVELS[logging.FATAL] = "FATAL"
DISABLE_TRACEBACKS = os.environ.get("DISABLE_TRACEBACKS")
class NullHandler(logging.Handler):
def emit(self, record):
pass
def get_logger(logger):
if isinstance(logger, basestring):
logger = logging.getLogger(logger)
if not logger.handlers:
logger.addHandler(NullHandler())
return logger
def anon_logger(name):
logger = logging.getLogger(name)
logger.addHandler(NullHandler())
return logger
def get_loglevel(level):
if isinstance(level, basestring):
return LOG_LEVELS[level]
return level
def naive_format_parts(fmt):
l = fmt.split('%')
for i, e in enumerate(l[1:]):
if not e or not l[i - 1]:
yield
elif e[0] in ["r", "s"]:
yield e[0]
def safeify_format(fmt, *args):
for index, type in enumerate(naive_format_parts(fmt)):
if not type:
yield args[index]
elif type == 'r':
yield safe_repr(args[index])
elif type == 's':
yield safe_str(args[index])
class LogMixin(object):
def debug(self, *args, **kwargs):
return self.log(logging.DEBUG, *args, **kwargs)
def info(self, *args, **kwargs):
return self.log(logging.INFO, *args, **kwargs)
def warn(self, *args, **kwargs):
return self.log(logging.WARN, *args, **kwargs)
def error(self, *args, **kwargs):
return self._error(logging.ERROR, *args, **kwargs)
def critical(self, *args, **kwargs):
return self._error(logging.CRITICAL, *args, **kwargs)
def _error(self, severity, *args, **kwargs):
kwargs.setdefault("exc_info", sys.exc_info())
if DISABLE_TRACEBACKS:
kwargs.pop("exc_info", None)
return self.log(severity, *args, **kwargs)
def annotate(self, text):
return "%s - %s" % (self.logger_name, text)
def log(self, severity, *args, **kwargs):
if self.logger.isEnabledFor(severity):
log = self.logger.log
if len(args) > 1 and isinstance(args[0], basestring):
expand = [maybe_promise(arg) for arg in args[1:]]
return log(severity,
self.annotate(args[0].replace('%r', '%s')),
*list(safeify_format(args[0], *expand)), **kwargs)
else:
return self.logger.log(severity,
self.annotate(" ".join(map(safe_str, args))),
**kwargs)
def get_logger(self):
return get_logger(self.logger_name)
def is_enabled_for(self, level):
return self.logger.isEnabledFor(self.get_loglevel(level))
def get_loglevel(self, level):
if not isinstance(level, int):
return LOG_LEVELS[level]
return level
@cached_property
def logger(self):
return self.get_logger()
@property
def logger_name(self):
return self.__class__.__name__
class Log(LogMixin):
def __init__(self, name, logger=None):
self._logger_name = name
self._logger = logger
def get_logger(self):
if self._logger:
return self._logger
return LogMixin.get_logger(self)
@property
def logger_name(self):
return self._logger_name
def setup_logging(loglevel=None, logfile=None):
logger = logging.getLogger()
loglevel = get_loglevel(loglevel or "ERROR")
logfile = logfile if logfile else sys.__stderr__
if not logger.handlers:
if hasattr(logfile, "write"):
handler = logging.StreamHandler(logfile)
else:
handler = WatchedFileHandler(logfile)
logger.addHandler(handler)
logger.setLevel(loglevel)
return logger
|