def setup_logging(logger, logfile=None, debug=False):
"""Configure a logger"""
fmt_str = "%(asctime)s: %(levelname)s:" \
" %(module)s.%(funcName)s():%(lineno)d: %(message)s"
console_handler = logging.StreamHandler(stream=sys.stdout)
console_handler.setFormatter(logging.Formatter(fmt_str))
logger.handlers = []
logger.addHandler(console_handler)
if logfile:
file_handler = WatchedFileHandler(logfile)
file_handler.setFormatter(logging.Formatter(fmt_str))
logger.addHandler(file_handler)
if debug:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
python类WatchedFileHandler()的实例源码
def register_loggers(app):
if app.debug:
return
import logging
from logging.handlers import WatchedFileHandler
import sys
logfile = app.config['DOORMAN_LOGGING_FILENAME']
if logfile == '-':
handler = logging.StreamHandler(sys.stdout)
else:
handler = WatchedFileHandler(logfile)
levelname = app.config['DOORMAN_LOGGING_LEVEL']
if levelname in ('DEBUG', 'INFO', 'WARN', 'WARNING', 'ERROR', 'CRITICAL'):
handler.setLevel(getattr(logging, levelname))
formatter = logging.Formatter(app.config['DOORMAN_LOGGING_FORMAT'])
handler.setFormatter(formatter)
app.logger.addHandler(handler)
def prepare_service_logger():
logger = logging.getLogger('TelegramBot')
logger.setLevel(logging.DEBUG)
logger.propagate = False
ch = WatchedFileHandler(filename=os.path.join(LOG_DIR, 'service.log'))
ch.setFormatter(ServiceClientFormatter(fmt="{asctime} | {action} | {method} {full_url} | {message}",
request_fmt="\nHeaders:\n{headers}\nBody:\n{body}",
response_fmt=" | {status_code} {status_text} | "
"{headers_elapsed}\nHeaders:\n{headers}\nBody:\n{body}",
exception_fmt=" | {exception_repr}",
parse_exception_fmt=" | {status_code} {status_text} | "
"{headers_elapsed} | {exception_repr}\nHeaders:\n"
"{headers}\nBody:\n{body}",
headers_fmt="\t{name}: {value}",
headers_sep="\n",
datefmt="%Y-%m-%dT%H:%M:%S%z",
style='{'))
ch.setLevel(logging.DEBUG)
logger.addHandler(ch)
return logger
def _add_file_handler(logger, filename=None, use_formatter=True):
if not _has_handler(logger, WatchedFileHandler):
if filename is None:
filename = _log_filename
handler = WatchedFileHandler(filename)
if use_formatter:
handler.setFormatter(logging.Formatter(FILE_FORMAT))
logger.addHandler(handler)
def __init__(self, filename, tz=timezone.utc):
self.loghandler = WatchedFileHandler(filename, encoding='utf-8', delay=True)
self.loghandler.setLevel(logging.INFO)
self.loghandler.setFormatter(logging.Formatter('%(message)s'))
self.tz = tz
def setupLogging(stream=None, path=None, level=None, format=None):
"""Setup logging.
Either a stream or a path can be provided. When a path is provided a log
handler that works correctly with C{logrotate} is used. Generally
speaking, C{stream} should only be used for non-file streams that don't
need log rotation.
@param name: The name of the logger to setup.
@param stream: The stream to write output to.
@param path: The path to write output to.
@param level: Optionally, the log level to set on the logger. Default is
C{logging.INFO}.
@param format: A format string for the logger.
@raise RuntimeError: Raised if neither C{stream} nor C{path} are provided,
or if both are provided.
@return: The configured logger, ready to use.
"""
if (not stream and not path) or (stream and path):
raise RuntimeError('A stream or path must be provided.')
if stream:
handler = StreamHandler(stream)
else:
handler = WatchedFileHandler(path)
if format is None:
format = '%(asctime)s %(levelname)8s %(message)s'
formatter = Formatter(format)
handler.setFormatter(formatter)
log = getLogger()
log.addHandler(handler)
log.propagate = False
log.setLevel(level or INFO)
return log
def get_handler(debug=False, syslog=False, logfile=None, rotation_parameters=None):
"""get an apropriate handler according to given parameters"""
if os.environ.get('APYCOT_ROOT'):
handler = logging.StreamHandler(sys.stdout)
if debug:
handler = logging.StreamHandler()
elif logfile is None:
if syslog:
from logging import handlers
handler = handlers.SysLogHandler()
else:
handler = logging.StreamHandler()
else:
try:
if rotation_parameters is None:
if os.name == 'posix' and sys.version_info >= (2, 6):
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler(logfile)
else:
handler = logging.FileHandler(logfile)
else:
from logging.handlers import TimedRotatingFileHandler
handler = TimedRotatingFileHandler(
logfile, **rotation_parameters)
except IOError:
handler = logging.StreamHandler()
return handler
def _setup_logger(self, level, log_file):
"""Setup log level and log file if set"""
if not level:
return
if logger.handlers:
return
if hasattr(logging, 'NullHandler'):
logger.addHandler(logging.NullHandler())
formatter = logging.Formatter(
'[%(levelname)s] %(asctime)s - %(module)s.%(funcName)s() '
'- %(message)s')
level = getattr(logging, level.upper())
logger.setLevel(level)
handler = logging.StreamHandler()
logger.addHandler(handler)
handler.setFormatter(formatter)
if not log_file:
return
try:
_handler = WatchedFileHandler(log_file)
except IOError:
logger.error("Could not write to %s, falling back to stdout",
log_file)
else:
logger.addHandler(_handler)
_handler.setFormatter(formatter)
def get_handler(debug=False, syslog=False, logfile=None, rotation_parameters=None):
"""get an apropriate handler according to given parameters"""
if os.environ.get('APYCOT_ROOT'):
handler = logging.StreamHandler(sys.stdout)
if debug:
handler = logging.StreamHandler()
elif logfile is None:
if syslog:
from logging import handlers
handler = handlers.SysLogHandler()
else:
handler = logging.StreamHandler()
else:
try:
if rotation_parameters is None:
if os.name == 'posix' and sys.version_info >= (2, 6):
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler(logfile)
else:
handler = logging.FileHandler(logfile)
else:
from logging.handlers import TimedRotatingFileHandler
handler = TimedRotatingFileHandler(
logfile, **rotation_parameters)
except IOError:
handler = logging.StreamHandler()
return handler
def get_handler(debug=False, syslog=False, logfile=None, rotation_parameters=None):
"""get an apropriate handler according to given parameters"""
if os.environ.get('APYCOT_ROOT'):
handler = logging.StreamHandler(sys.stdout)
if debug:
handler = logging.StreamHandler()
elif logfile is None:
if syslog:
from logging import handlers
handler = handlers.SysLogHandler()
else:
handler = logging.StreamHandler()
else:
try:
if rotation_parameters is None:
if os.name == 'posix' and sys.version_info >= (2, 6):
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler(logfile)
else:
handler = logging.FileHandler(logfile)
else:
from logging.handlers import TimedRotatingFileHandler
handler = TimedRotatingFileHandler(
logfile, **rotation_parameters)
except IOError:
handler = logging.StreamHandler()
return handler
def get_handler(debug=False, syslog=False, logfile=None, rotation_parameters=None):
"""get an apropriate handler according to given parameters"""
if os.environ.get('APYCOT_ROOT'):
handler = logging.StreamHandler(sys.stdout)
if debug:
handler = logging.StreamHandler()
elif logfile is None:
if syslog:
from logging import handlers
handler = handlers.SysLogHandler()
else:
handler = logging.StreamHandler()
else:
try:
if rotation_parameters is None:
if os.name == 'posix' and sys.version_info >= (2, 6):
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler(logfile)
else:
handler = logging.FileHandler(logfile)
else:
from logging.handlers import TimedRotatingFileHandler
handler = TimedRotatingFileHandler(
logfile, **rotation_parameters)
except IOError:
handler = logging.StreamHandler()
return handler
def init(self, *priv):
if not priv:
raise error.SnmpsimError('Bad log file params, need filename')
if sys.platform[:3] == 'win':
# fix possibly corrupted absolute windows path
if len(priv[0]) == 1 and priv[0].isalpha() and len(priv) > 1:
priv = [priv[0] + ':' + priv[1]] + list(priv[2:])
maxsize = 0
maxage = None
if len(priv) > 1 and priv[1]:
localtime = time.localtime()
if priv[1][-1] in ('k', 'K'):
maxsize = int(priv[1][:-1]) * 1024
elif priv[1][-1] in ('m', 'M'):
maxsize = int(priv[1][:-1]) * 1024 * 1024
elif priv[1][-1] in ('g', 'G'):
maxsize = int(priv[1][:-1]) * 1024 * 1024 * 1024
elif priv[1][-1] in ('h', 'H'):
maxage = ('H', int(priv[1][:-1]))
elif priv[1][-1] in ('d', 'D'):
maxage = ('D', int(priv[1][:-1]))
else:
raise error.SnmpsimError(
'Unknown log rotation criteria %s, use <NNN>K,M,G for size or <NNN>H,D for time limits' % priv[1]
)
try:
if maxsize:
handler = handlers.RotatingFileHandler(priv[0], backupCount=30, maxBytes=maxsize)
elif maxage:
handler = handlers.TimedRotatingFileHandler(priv[0], backupCount=30, when=maxage[0], interval=maxage[1])
else:
handler = handlers.WatchedFileHandler(priv[0])
except AttributeError:
raise error.SnmpsimError(
'Bad log rotation criteria: %s' % sys.exc_info()[1]
)
handler.setFormatter(logging.Formatter('%(asctime)s %(name)s: %(message)s'))
self._logger.addHandler(handler)
self('Log file %s, rotation rules: %s' % (
priv[0], maxsize and '> %sKB' % (maxsize / 1024) or maxage and '%s%s' % (maxage[1], maxage[0]) or '<none>'))
