Source code for foundations.verbose

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
**verbose.py**

**Platform:**
	Windows, Linux, Mac Os X.

**Description:**
	Defines **Foundations** package verbose and logging objects.

**Others:**

"""

#**********************************************************************************************************************
#***	Future imports.
#**********************************************************************************************************************
from __future__ import unicode_literals

#**********************************************************************************************************************
#***	External imports.
#**********************************************************************************************************************
import functools
import hashlib
import inspect
import itertools
import logging
import sys
import threading
import tempfile

#**********************************************************************************************************************
#***	Internal imports.
#**********************************************************************************************************************
import foundations.trace
from foundations.globals.constants import Constants

#**********************************************************************************************************************
#***	Module attributes.
#**********************************************************************************************************************
__author__ = "Thomas Mansencal"
__copyright__ = "Copyright (C) 2008 - 2014 - Thomas Mansencal"
__license__ = "GPL V3.0 - http://www.gnu.org/licenses/"
__maintainer__ = "Thomas Mansencal"
__email__ = "thomas.mansencal@gmail.com"
__status__ = "Production"

__all__ = ["THREADS_IDENTIFIERS",
		   "INDENT_LEVEL",
		   "toString",
		   "LOGGER",
		   "LOGGING_DEFAULT_FORMATTER",
		   "LOGGING_EXTENDED_FORMATTER",
		   "LOGGING_STANDARD_FORMATTER",
		   "TRACER_LOGGING_FUNCTION",
		   "Streamer"
		   "StandardOutputStreamer",
		   "indentMessage",
		   "tracer",
		   "installLogger",
		   "uninstallLogger",
		   "getLoggingConsoleHandler",
		   "getLoggingFileHandler",
		   "getLoggingStreamHandler",
		   "removeLoggingHandler",
		   "setVerbosityLevel"]

THREADS_IDENTIFIERS = {}

INDENT_LEVEL = 0

#**********************************************************************************************************************
#***	Module attributes.
#**********************************************************************************************************************
[docs]def toUnicode(data, encoding=Constants.defaultCodec, errors=Constants.codecError): """ Converts given data to unicode string using package default settings, fighting **The Hell**! Usage:: >>> toUnicode("myData") u'myData' >>> toUnicode("汉字/漢字") u'\u6c49\u5b57/\u6f22\u5b57' :param data: Data to convert. :type data: object :param encoding: File encoding codec. :type encoding: unicode :param errors: File encoding errors handling. :type errors: unicode :return: Unicode data. :rtype: unicode """ if isinstance(data, type("")): return data else: try: return unicode(data, encoding, errors) except TypeError: return unicode(str(data), encoding, errors) #********************************************************************************************************************** #*** Logging classes and definitions monkey patching. #**********************************************************************************************************************
def _LogRecord_getAttribute(self, attribute): """ Overrides logging.LogRecord.__getattribute__ method in order to manipulate requested attributes values. :param attribute: Attribute name. :type attribute: unicode :return: Modified method. :rtype: object """ if attribute == "__dict__": threadIdent = threading.currentThread().ident if not threadIdent in THREADS_IDENTIFIERS: THREADS_IDENTIFIERS[threadIdent] = (threading.currentThread().name, hashlib.md5(threading.currentThread().name).hexdigest()[:8]) object.__getattribute__(self, attribute)["threadName"] = THREADS_IDENTIFIERS[threadIdent][1] return object.__getattribute__(self, attribute) else: return object.__getattribute__(self, attribute) logging.LogRecord.__getattribute__ = _LogRecord_getAttribute def _LogRecord_msg(): """ Overrides logging.LogRecord.msg attribute to ensure variable content is stored as unicode. """ def _LogRecord_msgProperty(self): return self.__msg def _LogRecord_msgSetter(self, value): self.__msg = toUnicode(value) logging.LogRecord.msg = property(_LogRecord_msgProperty, _LogRecord_msgSetter) _LogRecord_msg() #********************************************************************************************************************** #*** Module attributes. #********************************************************************************************************************** LOGGER = logging.getLogger(Constants.logger) LOGGING_DEFAULT_FORMATTER = logging.Formatter("%(levelname)-8s: %(message)s") LOGGING_EXTENDED_FORMATTER = logging.Formatter("%(asctime)s - %(threadName)s - %(levelname)-8s: %(message)s") LOGGING_STANDARD_FORMATTER = logging.Formatter() TRACER_LOGGING_FUNCTION = LOGGER.info #********************************************************************************************************************** #*** Module classes and definitions. #**********************************************************************************************************************
[docs]class Streamer(object): """ Defines a stream object for :class:`logging.StreamHandler` logging handler. """ def __init__(self, stream=None): """ Initializes the class. :param stream: Stream object. :type stream: object """ self.__stream = [] #****************************************************************************************************************** #*** Attributes properties. #****************************************************************************************************************** @property def stream(self): """ Property for **self.__stream** attribute. :return: self.__stream. :rtype: list """ return self.__stream @stream.setter def stream(self, value): """ Setter for **self.__stream** attribute. :param value: Attribute value. :type value: list """ if value is not None: assert type(value) is list, "'{0}' attribute: '{1}' type is not 'list'!".format("stream", value) self.__stream = value @stream.deleter
[docs] def stream(self): """ Deleter for **self.__stream** attribute. """ raise Exception("{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "stream")) #****************************************************************************************************************** #*** Class methods. #******************************************************************************************************************
[docs] def write(self, message): """ Provides write ability to the class. :param message: Current message. :type message: unicode """ self.__stream.append(message)
[docs] def flush(self): """ Flushes the current stream. """ pass
[docs]class StandardOutputStreamer(object): """ | Defines a redirection object intented to be used for :data:`sys.stdout` and :data:`sys.stderr` streams. | Logging messages will be written to given logger handlers. """ def __init__(self, logger): """ Initializes the class. :param logger: Logger. :type logger: object """ self.__logger = logger #****************************************************************************************************************** #*** Attributes properties. #****************************************************************************************************************** @property def logger(self): """ Property for **self.__logger** attribute. :return: self.__logger. :rtype: Logger """ return self.__logger @logger.setter def logger(self, value): """ Setter for **self.__logger** attribute. :param value: Attribute value. :type value: Logger """ raise Exception("{0} | '{1}' attribute is read only!".format(self.__class__.__name__, "logger")) @logger.deleter
[docs] def logger(self): """ Deleter for **self.__logger** attribute. """ raise Exception("{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "logger")) #****************************************************************************************************************** #*** Class methods. #******************************************************************************************************************
[docs] def write(self, message): """ Writes given message to logger handlers. :param message: Message. :type message: unicode :return: Method success. :rtype: bool """ for handler in self.__logger.__dict__["handlers"]: handler.stream.write(message) return True
[docs]def indentMessage(message): """ Idents given message using the attr`INDENT_LEVEL` attribute value. :param message: Message to indent. :type message: unicode :return: indented message. :rtype: unicode """ return "{0}{1}".format(" " * 4 * INDENT_LEVEL, message)
[docs]def tracer(object): """ | Traces execution. | Any method / definition decorated will have it's execution traced through debug messages. | Both object entry and exit are logged. Entering in an object:: INFO : ---> foundations.environment.getUserApplicationDataDirectory() <<<--- Exiting from an object:: INFO : <--- foundations.environment.getSystemApplicationDataDirectory() ^ '...' ---> :param object: Object to decorate. :type object: object :return: Object. :rtype: object """ # Oncilla: Statement commented by auto-documentation process: # Oncilla: Statement commented by auto-documentation process: @functools.wraps(object) # Oncilla: Statement commented by auto-documentation process: @functools.partial(foundations.trace.validateTracer, object) def tracerWrapper(*args, **kwargs): """ Traces execution. :param \*args: Arguments. :type \*args: \* :param \*\*kwargs: Keywords arguments. :type \*\*kwargs: \*\* :return: Object. :rtype: object """ global INDENT_LEVEL traceName = foundations.trace.getTraceName(object) code = object.func_code argsCount = code.co_argcount argsNames = code.co_varnames[:argsCount] functionDefaults = object.func_defaults or list() argsDefaults = dict(zip(argsNames[-len(functionDefaults):], functionDefaults)) positionalArgs = map(foundations.trace.formatArgument, zip(argsNames, args)) defaultedArgs = [foundations.trace.formatArgument((name, argsDefaults[name])) \ for name in argsNames[len(args):] if name not in kwargs] namelessArgs = map(repr, args[argsCount:]) keywordArgs = map(foundations.trace.formatArgument, kwargs.items()) TRACER_LOGGING_FUNCTION(indentMessage("---> {0}({1}) <---".format(traceName, ", ".join(itertools.chain(positionalArgs, defaultedArgs, namelessArgs, keywordArgs))))) INDENT_LEVEL += 1 value = object(*args, **kwargs) INDENT_LEVEL -= 1 TRACER_LOGGING_FUNCTION(indentMessage("<--- {0} ^ {1} --->".format(traceName, repr(value)))) return value return tracerWrapper
[docs]def installLogger(logger=None, module=None): """ Installs given logger in given module or default logger in caller introspected module. :param logger: Logger to install. :type logger: Logger :param module: Module. :type module: ModuleType :return: Logger. :rtype: Logger """ logger = logging.getLogger(Constants.logger) if logger is None else logger if module is None: # Note: inspect.getmodule() can return the wrong module if it has been imported with different relatives paths. module = sys.modules.get(inspect.currentframe().f_back.f_globals["__name__"]) setattr(module, "LOGGER", logger) foundations.trace.registerModule(module) return logger
[docs]def uninstallLogger(logger=None, module=None): """ Uninstalls given logger in given module or default logger in caller introspected module. :param logger: Logger to uninstall. :type logger: Logger :param module: Module. :type module: ModuleType :return: Definition success. :rtype: bool """ logger = logging.getLogger(Constants.logger) if logger is None else logger if module is None: # Note: inspect.getmodule() can return the wrong module if it has been imported with different relatives paths. module = sys.modules.get(inspect.currentframe().f_back.f_globals["__name__"]) hasattr(module, "LOGGER") and delattr(module, "LOGGER") return True
[docs]def getLoggingConsoleHandler(logger=None, formatter=LOGGING_DEFAULT_FORMATTER): """ Adds a logging console handler to given logger or default logger. :param logger: Logger to add the handler to. :type logger: Logger :param formatter: Handler formatter. :type formatter: Formatter :return: Added handler. :rtype: Handler """ logger = LOGGER if logger is None else logger loggingConsoleHandler = logging.StreamHandler(sys.__stdout__) loggingConsoleHandler.setFormatter(formatter) logger.addHandler(loggingConsoleHandler) return loggingConsoleHandler
[docs]def getLoggingFileHandler(logger=None, file=None, formatter=LOGGING_DEFAULT_FORMATTER): """ Adds a logging file handler to given logger or default logger using given file. :param logger: Logger to add the handler to. :type logger: Logger :param file: File to verbose into. :type file: unicode :param formatter: Handler formatter. :type formatter: Formatter :return: Added handler. :rtype: Handler """ logger = LOGGER if logger is None else logger file = tempfile.NamedTemporaryFile().name if file is None else file loggingFileHandler = logging.FileHandler(file) loggingFileHandler.setFormatter(formatter) logger.addHandler(loggingFileHandler) return loggingFileHandler
[docs]def getLoggingStreamHandler(logger=None, formatter=LOGGING_DEFAULT_FORMATTER): """ Adds a logging stream handler to given logger or default logger using given file. :param logger: Logger to add the handler to. :type logger: Logger :param file: File to verbose into. :type file: unicode :param formatter: Handler formatter. :type formatter: Formatter :return: Added handler. :rtype: Handler """ logger = LOGGER if logger is None else logger loggingStreamHandler = logging.StreamHandler(Streamer()) loggingStreamHandler.setFormatter(formatter) logger.addHandler(loggingStreamHandler) return loggingStreamHandler
[docs]def removeLoggingHandler(handler, logger=None): """ Removes given logging handler from given logger. :param handler: Handler. :type handler: Handler :param logger: Handler logger. :type logger: Logger :return: Definition success. :rtype: bool """ logger = LOGGER if logger is None else logger logger.handlers and LOGGER.debug("> Stopping handler: '{0}'.".format(handler)) logger.removeHandler(handler) return True
[docs]def setVerbosityLevel(verbosityLevel=3, logger=None): """ Defines logging verbosity level. Available verbosity levels:: 0: Critical. 1: Error. 2: Warning. 3: Info. 4: Debug. :param verbosityLevel: Verbosity level. :type verbosityLevel: int :param logger: Logger to set the verbosity level to. :type logger: Logger :return: Definition success. :rtype: bool """ logger = LOGGER if logger is None else logger if verbosityLevel == 0: logger.setLevel(logging.CRITICAL) logging.disable(logging.ERROR) elif verbosityLevel == 1: logger.setLevel(logging.ERROR) logging.disable(logging.WARNING) elif verbosityLevel == 2: logger.setLevel(logging.WARNING) logging.disable(logging.INFO) elif verbosityLevel == 3: logger.setLevel(logging.INFO) logging.disable(logging.DEBUG) elif verbosityLevel == 4: logger.setLevel(logging.DEBUG) logging.disable(logging.NOTSET) return True