#!/usr/bin/env python
# -*- coding: utf-8 -*-
# created at 27.03.2011
# This software is distributed under the BSD license.
# http://www.opensource.org/licenses/bsd-license.php
# Copyright © 2011, Oleg Churkin
"""
Here is a wrapper for standard :mod:`logging` module, which provides some new
logging facilities and allows accessing them in more Pythonic way.
"""
__version__ = "0.1"
import sys
import inspect
# Python version check: 2.6 and higher is required.
if (sys.version_info[0] < 2 or
(sys.version_info[0] == 2 and sys.version_info[1] < 6)):
raise ImportError("Minimum Python version 2.6 is required."
" Your current Python version is"
" %s.%s." % sys.version_info[:2])
import logging
import functools
from logging import (INFO, DEBUG, CRITICAL, WARN, #IGNORE:W0611 @UnusedImport
WARNING, ERROR, FATAL, NOTSET) #@UnusedImport
__all__ = ["INFO", "DEBUG", "CRITICAL", "WARN", "WARNING", "ERROR", "FATAL",
"NOTSET", "DOC", "STDOUT", "STDERR", "log", "info", "debug", "error",
"warning", "warn","exception", "critical", "compose_handler",
"get_logger", "get_default_logger", "LoggingOptions", "LoggerAdapter"]
# logging level priorities:
# NOTSET < DEBUG < DOC < INFO < STDOUT < WARNING (WARN) < ERROR < STDERR < CRITICAL (FATAL)
DOC = 15
logging.addLevelName(DOC, "DOC")
STDOUT = 25
logging.addLevelName(STDOUT, "STDOUT")
STDERR = 45
logging.addLevelName(STDERR, "STDERR")
class Defaults(object): #IGNORE:R0903
"""Namespace with default values."""
GLOBAL_LOGGING_LEVEL = DEBUG
STREAM_MESSAGE_FORMAT = ("%(asctime)s [%(levelname)-8s] %(message)s", "%H:%M:%S")
FILE_MESSAGE_FORMAT = "%(asctime)s [%(levelname)-8s] %(message)s"
[docs]class LoggingOptions(object): #IGNORE:R0903
"""This structure contains different logging options.
Available options:
* **indent_size** - specifies number of spaces to be inserted before
'%(message)s' parameter, see logging.Formatter for more details;
* **enable_new_style** - if ``True`` log's methods will join all provided
positional arguments with space as delimiter, e.g.:
>>> log.info("I", "am", "fine.")
I am fine.
* **new_style_delimiter** - delimiter between message parts when new
message style is enabled, space by default;
* **enable_decorator_mode** - if ``True`` log's methods will produce necessary
messages when used as decorators, otherwise no output will be produced;
* **enable_doc_processing** - if ``True`` all doc strings will be logged by log
or its method used as decorators;
* **doc_processing_prefix** - if specified, only doc strings begins with
this character(s) will be logged. Character(s) will be stripped
from logged message;
* **doc_processing_level** - logging level for logged doc strings, by default
it's :const:`DOC`;
"""
indent = None
enable_new_style = True
new_style_delimiter = " "
enable_decorator_mode = True
enable_doc_processing = True
doc_processing_prefix = None
doc_processing_level = DOC
[docs]def compose_handler(handler, level=None, format_tuple=None, filter_name=None):
"""Help to compose final handler.
:param handler: any handler object from logging and logging.handlers;
:param level: logging level (``INFO``, ``DEBUG`` and so on);
:param format_tuple: message format tuple (fmt, datefmt), see :class:`logging.Formatter`.
Only if string is provided, ``datefmt`` will be passed as None;
:param filter_name: filter name, see :class:`logging.Filter`;
"""
if level is not None:
handler.setLevel(level)
if format_tuple is not None:
if isinstance(format_tuple, basestring):
format_tuple = (format_tuple, None)
handler.setFormatter(logging.Formatter(fmt=format_tuple[0],
datefmt=format_tuple[1]))
if filter_name is not None:
handler.addFilter(logging.Filter(name=filter_name))
return handler
[docs]def get_logger(name=None, level=None, handlers=None, options=None, extra=None):
"""Return basic logger object.
:param name: logger's name;
:param level: logging level, if not specified and some handlers are provided,
minimum of all their levels will be taken. If final value is ``NOTSET``,
then :const:`Defaults.GLOBAL_LOGGING_LEVEL` will be used;
:param handlers: list of handlers for created logger to attach,
if value is not iterable, it will be put in
a list automatically;
:param options: :class:`LoggingOptions` object;
:param extra: extra arguments for underlying logging functionality, see
documentation for :mod:`logging`;
"""
if level is None:
level = min([_.level for _ in (handlers or [])] or [NOTSET])
if level == NOTSET:
level = Defaults.GLOBAL_LOGGING_LEVEL
if name:
logger = logging.getLogger(name)
logger.setLevel(level)
else:
logger = logging.RootLogger(level)
if handlers is not None:
if not isinstance(handlers, (list, tuple)):
handlers = [handlers]
for handler in handlers:
logger.addHandler(handler)
return LoggerAdapter(logger, options, extra)
[docs]def get_default_logger(stream=None):
"""Return default logger for this module.
:param stream: stream object for default logging handler :class:`logging.StreamHandler`,
default value is ``sys.__stdout__``.
"""
if stream is None:
stream = sys.__stdout__
stream_handler = compose_handler(logging.StreamHandler(stream),
Defaults.GLOBAL_LOGGING_LEVEL,
Defaults.STREAM_MESSAGE_FORMAT)
return get_logger(handlers=[stream_handler])
# this function copied from inspect module in Python 2.7, the algorithm was
# slightly changed: now it returns list of tuples:
# [(name, value), ...] to retain sequence of attributes.
# Ironically ordered dictionary can be used here, since it was added in 2.7 only.
def getcallargs(func, *positional, **named): #IGNORE:R0912 #IGNORE:R0914
"""Get the mapping of arguments to values.
A dict is returned, with keys the function argument names (including the
names of the * and ** arguments, if any), and values the respective bound
values from 'positional' and 'named'."""
args, varargs, varkw, defaults = inspect.getargspec(func)
f_name = func.__name__
arg2value_keys = []
arg2value_vals = []
# The following closures are basically because of tuple parameter unpacking.
assigned_tuple_params = []
def assign(arg, value): #IGNORE:C0111
if isinstance(arg, str):
arg2value_keys.append(arg)
arg2value_vals.append(value)
else:
assigned_tuple_params.append(arg)
value = iter(value)
for i, subarg in enumerate(arg):
try:
subvalue = next(value)
except StopIteration:
raise ValueError('need more than %d %s to unpack' %
(i, 'values' if i > 1 else 'value'))
assign(subarg, subvalue)
try:
next(value)
except StopIteration:
pass
else:
raise ValueError('too many values to unpack')
def is_assigned(arg): #IGNORE:C0111
if isinstance(arg, str):
return arg in arg2value_keys
return arg in assigned_tuple_params
if inspect.ismethod(func) and func.im_self is not None:
# implicit 'self' (or 'cls' for classmethods) argument
positional = (func.im_self,) + positional
num_pos = len(positional)
num_total = num_pos + len(named)
num_args = len(args)
num_defaults = len(defaults) if defaults else 0
for arg, value in zip(args, positional):
assign(arg, value)
if varargs:
if num_pos > num_args:
assign(varargs, positional[-(num_pos-num_args):])
else:
assign(varargs, ())
elif 0 < num_args < num_pos:
raise TypeError('%s() takes %s %d %s (%d given)' % (
f_name, 'at most' if defaults else 'exactly', num_args,
'arguments' if num_args > 1 else 'argument', num_total))
elif num_args == 0 and num_total:
raise TypeError('%s() takes no arguments (%d given)' %
(f_name, num_total))
for arg in args:
if isinstance(arg, str) and arg in named:
if is_assigned(arg):
raise TypeError("%s() got multiple values for keyword "
"argument '%s'" % (f_name, arg))
else:
assign(arg, named.pop(arg))
if defaults: # fill in any missing values with the defaults
for arg, value in zip(args[-num_defaults:], defaults):
if not is_assigned(arg):
assign(arg, value)
if varkw:
assign(varkw, named)
elif named:
unexpected = next(iter(named))
if isinstance(unexpected, unicode):
unexpected = unexpected.encode(sys.getdefaultencoding(), 'replace')
raise TypeError("%s() got an unexpected keyword argument '%s'" %
(f_name, unexpected))
unassigned = num_args - len([arg for arg in args if is_assigned(arg)])
if unassigned:
num_required = num_args - num_defaults
raise TypeError('%s() takes %s %d %s (%d given)' % (
f_name, 'at least' if defaults else 'exactly', num_required,
'arguments' if num_required > 1 else 'argument', num_total))
return zip(arg2value_keys, arg2value_vals)
class _LoggerDecoratorHelper(object): #IGNORE:R0903
"""Logger helper."""
def __init__(self, logger, level):
self.logger = logger
self.level = level
self.options = self.logger.options
self._cache = None
def _get_arguments(self, func, args, kwargs):
"""Return arguments and their values provided for function 'func'.
This information should be cached for current session. Probably
cache should be used for every equal triple (func, args, kwargs).
"""
if self._cache is None:
self._cache = getcallargs(func, *args, **kwargs)
return self._cache
def _parameters_to_string(self, func, args, kwargs): #IGNORE:R0201
"""Format provided arguments to output them as Python code."""
items = self._get_arguments(func, args, kwargs)
parent = dict(items).pop("self", None)
_ = []
for k, v in items:
if k != "self":
_.append("{0}={1}".format(k, "'{0}'".format(v) if isinstance(v, basestring) else v))
if parent:
# it's a method
# support for classmethods
if not inspect.isclass(parent):
parent = parent.__class__
func_info = "method {0}.{1}".format(parent.__name__,
func.__name__)
else:
# it's a function
func_info = "function {0}".format(func.__name__)
return "Executing {0}({1}).".format(func_info, ", ".join(_))
def _doc_processing(self, func, args, kwargs):
"""Log doc stings."""
doc = inspect.getdoc(func)
if doc is not None:
# formatting
try:
doc = doc.format(**dict(self._get_arguments(func, args, kwargs)))
except KeyError:
pass
for _ in doc.split("\n"):
if self.options.doc_processing_prefix:
if _.startswith(self.options.doc_processing_prefix):
_ = _[len(self.options.doc_processing_prefix):]
else:
continue
self.logger.log(self.options.doc_processing_level, _)
def __call__(self, func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
""" Simple wrapper."""
if self.options.enable_decorator_mode:
self.logger.log(self.level,
self._parameters_to_string(func, args, kwargs))
# log doc-strings
if self.options.enable_doc_processing:
self._doc_processing(func, args, kwargs)
# empty cached values:
self._cache = None
# execute the function
return func(*args, **kwargs)
return wrapper
class _LoggerStreamHelper(object):
"""Imitation of stream object.
sys.stdout (sys.stderr) can be replaced with this imitation.
Supported methods:
write, writelines, close, and flush
"""
def __init__(self, logger, level):
self.logger = logger
self.level = level
self.closed = False
self._cache = []
def write(self, msg):
"""Write provided string to a stream."""
if not self.closed:
if not msg.endswith("\n"):
self._cache.append(msg)
else:
self._cache.append(msg.strip("\n"))
self.flush()
def writelines(self, lines):
"""Write a sequence of strings to the file."""
for line in lines:
self.write(line)
def flush(self):
"""Flush cached messages to logger."""
if self._cache:
self.logger.log(self.level, "".join(self._cache))
self._cache = []
def close(self):
"""Close current stream."""
self.closed = True
# Copied from logging module and updated.
[docs]class LoggerAdapter(object):
"""
An adapter for loggers which makes it easier to specify contextual
information in logging output.
"""
def __init__(self, logger=None, options=None, extra=None):
"""
Initialize the adapter with a logger and a dict-like object which
provides contextual information. This constructor signature allows
easy stacking of LoggerAdapters, if so desired.
You can effectively pass keyword arguments as shown in the
following example:
adapter = LoggerAdapter(someLogger, extra=dict(p1=v1, p2="v2"))
"""
self.logger = logger
self.extra = extra or {}
self.options = options or LoggingOptions()
[docs] def process_message(self, msg, args, kwargs):
"""Process the logging message and keyword arguments passed in to
a logging call to insert contextual information. You can either
manipulate the message itself, args or keyword args. Return
the message, args and kwargs modified (or not) to suit your needs.
"""
self.extra.update(kwargs.get("extra", {}))
kwargs["extra"] = self.extra
# processing with 'indent' options
if self.options.indent is not None:
if isinstance(self.options.indent, int):
indent = " " * self.options.indent
else:
indent = self.options.indent
msg = "{0}{1}".format(indent, msg)
# processing with new message system
if self.options.enable_new_style:
# TODO: support new formatting style
args = (msg, ) + args
msg = self.options.new_style_delimiter.join(["%s" for _ in args])
return msg, args, kwargs
[docs] def debug(self, msg, *args, **kwargs):
"""
Delegate a debug call to the underlying logger, after adding
contextual information from this adapter instance.
"""
return self.log(DEBUG, msg, *args, **kwargs)
[docs] def info(self, msg, *args, **kwargs):
"""
Delegate an info call to the underlying logger, after adding
contextual information from this adapter instance.
"""
return self.log(INFO, msg, *args, **kwargs)
[docs] def warning(self, msg, *args, **kwargs):
"""
Delegate a warning call to the underlying logger, after adding
contextual information from this adapter instance.
"""
return self.log(WARNING, msg, *args, **kwargs)
[docs] def error(self, msg, *args, **kwargs):
"""
Delegate an error call to the underlying logger, after adding
contextual information from this adapter instance.
"""
return self.log(ERROR, msg, *args, **kwargs)
[docs] def exception(self, msg, *args, **kwargs):
"""
Delegate an exception call to the underlying logger, after adding
contextual information from this adapter instance.
"""
kwargs["exc_info"] = 1
return self.log(ERROR, msg, *args, **kwargs)
[docs] def critical(self, msg, *args, **kwargs):
"""
Delegate a critical call to the underlying logger, after adding
contextual information from this adapter instance.
"""
return self.log(CRITICAL, msg, *args, **kwargs)
[docs] def log(self, level, msg, *args, **kwargs):
"""
Delegate a log call to the underlying logger, after adding
contextual information from this adapter instance.
"""
# we need to detect if log's method used as decorators, if so msg
# should be a function (or method) and args, kwargs must be empty.
if ((inspect.ismethod(msg) or inspect.isfunction(msg)) and
not args and
(not kwargs or ("exc_info" in kwargs and len(kwargs.keys())==1))):
# ok, we are in 'decorator' mode
return _LoggerDecoratorHelper(self, level)(msg)
msg, args, kwargs = self.process_message(msg, args, kwargs)
self.logger.log(level, msg, *args, **kwargs)
[docs] def append_handlers(self, *handlers):
"""Append provided handlers to the list of existed ones."""
for handler in handlers:
self.logger.addHandler(handler)
[docs] def replace_handlers(self, *handlers):
"""Replace current list of handlers with the provided ones."""
for handler in self.logger.handlers:
self.logger.removeHandler(handler)
self.append_handlers(*handlers)
[docs] def set_level(self, level):
"""Set logging level for current logger only."""
self.logger.setLevel(level)
[docs] def get_level(self):
"""Return current logger's level."""
return self.logger.level
level = property(get_level, set_level)
[docs] def set_effective_level(self, level):
"""Set effective level: set the same logging level for current logger
and for all its handlers.
"""
for handler in self.logger.handlers:
handler.setLevel(level)
self.set_level(level)
[docs] def get_effective_level(self):
"""Return effective logging level, any message produced with this level
will be logged by all attached handlers.
"""
return max([_.level for _ in self.logger.handlers] + [self.logger.level])
effective_level = property(get_effective_level, set_effective_level)
[docs] def set_options(self, *args, **kwargs):
"""Set logging options, see :class:`LoggingOptions` class above.
:param args: here can be only one positional argument: instance of
LoggingOptions object, if it's provided, keyword arguments will
be ignored and current logging options will be replaced with
provided ones;
:param kwargs: any existing options can be provided (see available in
:class:`LoggingOptions` class);
"""
if args:
if len(args) != 1:
raise RuntimeError("Only one positional argument is allowed.")
if not isinstance(args[0], LoggingOptions):
raise TypeError("Provided argument must be an instance of"
" LoggingOptions object.")
# The interesting moment to notice here. If we perform a direct
# options assignment, we broke the link between self.options and
# self.logger.options passed to _LoggerDecoratorHelper, when
# decorator was created, thus any further call to log.set_options
# method will not change already created decorators behavior.
# So we need to set new properties one by one using 'setattr'.
# self.options = args[0]
kwargs = args[0].__dict__
# process keyword arguments
for key, value in kwargs.iteritems():
if not hasattr(self.options, key):
raise AttributeError("Unsupported option is passed: {0}"
.format(key))
setattr(self.options, key, value)
[docs] def get_stream(self, level=None):
"""Return stream imitation object, which can replace stdout and
stderr streams.
:param level: logging level, messages will be logged with,
usually :const:`STDOUT` or :const:`STDERR`;
"""
return _LoggerStreamHelper(self, level or self.level)
def __call__(self, msg, *args, **kwargs):
"""
Calling log instance should act as usual 'info' or 'debug' methods.
"""
return self.log(self.level, msg, *args, **kwargs)
# Default logger object, can be used right after it is imported.
log = get_default_logger()
[docs]def info(msg, *args, **kwargs):
"""Log a message with severity ``INFO`` on the root logger."""
return log.info(msg, *args, **kwargs)
[docs]def debug(msg, *args, **kwargs):
"""Log a message with severity ``DEBUG`` on the root logger."""
return log.debug(msg, *args, **kwargs)
[docs]def warning(msg, *args, **kwargs):
"""Log a message with severity ``WARNING`` on the root logger."""
return log.warning(msg, *args, **kwargs)
warn = warning
[docs]def error(msg, *args, **kwargs):
"""Log a message with severity ``ERROR`` on the root logger."""
return log.error(msg, *args, **kwargs)
[docs]def exception(msg, *args, **kwargs):
"""Log a message with severity ``ERROR`` on the root logger."""
return log.exception(msg, *args, **kwargs)
[docs]def critical(msg, *args, **kwargs):
"""Log a message with severity ``CRITICAL`` on the root logger."""
return log.critical(msg, *args, **kwargs)