Telegram API Question posted in
A comprehensive official documentation can be found here.

How to wrap python logging module? – Telegram API

I would like to wrap the creation of a logger and all its associated configuration, formatters and handlers in a separate module, so that from my application I would just write:

import my_logger

test_logger = my_logger.getlogger("logger_name")

This logger will be used in a unit testing framework using unittest.TestCase for logging purposes.

I use a QueueHandler and a QueueListener on one handler (as per https://docs.python.org/3/howto/logging-cookbook.html#dealing-with-handlers-that-block), as the handler is using the Telegram API and I would like to avoid hanging the whole application while waiting for the Telegram server to be available (for whatever reason).

The (perhaps silly) question is:

how can I automatically handle the start and stop of the QueueListener when I start and stop the execution of the tests? Do I have to subclass a class of the logging module (which one?) to add, say, a start/stop method?

Thank you very much.

UPDATED (01)

I think I was not complete in my question. I already have the wrapper for the logger, you will find it below:

import logging
from logging.handlers import QueueHandler
from logging.handlers import QueueListener
import queue
from time import strftime

from notifiers.logging import NotificationHandler


def getLogger(name):
            
    LOG_DIR = "logs"

    _test_logger = logging.getLogger(name)
    _test_logger.setLevel(logging.INFO)
    

    # Create a file handler for the test logger.
    # This handler will create the log file of the tests.
    # TODO: create the LOG_DIR if it does not exist
    now = strftime("%Y%m%d_%H%M%S")
    # TODO: use os.path methods instead of directly writing path into filenames
    _tl_fh = logging.FileHandler("..\{}\{}_{}.txt".format(LOG_DIR, now, name))
    _tl_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
    _tl_fh.setFormatter(_tl_formatter)
    _test_logger.addHandler(_tl_fh)
    
    # Create a console handler for the test logger.
    # This is the logger to the console
    _tl_console = logging.StreamHandler()
    _test_logger.addHandler(_tl_console)
    
    # Create a notifier logger. This will notify via Telegram:
    # * start of a test suite
    # * Stop of a test suite
    # * Test suite and test case errors
    
    # Create an unlimited size queue and attach it to a queue handler
    que = queue.Queue(-1)
    queue_handler = QueueHandler(que)
    # Create a Telegram notification handler.  
    hdlr = NotificationHandler("telegram", defaults=defaults)
    # On the other side of the queue attach a queue listener
    _listener = QueueListener(que, hdlr)
    queue_handler.setLevel(logging.ERROR)
    _test_logger.addHandler(queue_handler)
    formatter = logging.Formatter("%(asctime)s - %(levelno)s - %(levelname)s - %(message)s")

    hdlr.setFormatter(formatter)
    #=======================================================================
    # Start the queue listener
    #=======================================================================
    
    _listener.start()
    
    return _test_logger

The "problem" is the _listener.start() line. How can I stop the listener at the end of the test?
Is it possible to add a start/stop methods to the loggers, so that I can write something like?

import my_logger

test_logger = my_logger.getlogger("logger_name")
[...] execute some test here

test_logger.stop()
Tags:

4

Answers


  1. Chosen as BEST ANSWER
    0

    I think I have found a solution, maybe not the most pythonic but it fit my need. I found the following answer (python logging - With JSON logs can I add an "extra" value to every single log?)

    Following the proposed answer, I wrapped the creation and configuration of the logger into a class named TestLogger and I added the logic I needed, i.e. the start_logger() and stop_logger().

    What I do not like is that I have to redefine the various methods debug(), info(), warning(), error(), just to make them available. If someone found a better solution please let me know.

    Below the complete code.

    import logging
from logging.handlers import QueueHandler
from logging.handlers import QueueListener
import queue
from time import strftime

from notifiers.logging import NotificationHandler


class TestLogger(object):

    def __init__(self, logger_name):

        LOG_DIR = "logs"

        self._test_logger = logging.getLogger(logger_name)
        self._test_logger.setLevel(logging.INFO)


        # Create a file handler for the test logger.
        # This handler will create the log file of the tests.
        # TODO: create the LOG_DIR if it does not exist
        now = strftime("%Y%m%d_%H%M%S")
        # TODO: use os.path methods instead of directly writing path into filenames
        _tl_fh = logging.FileHandler("..\{}\{}_{}.txt".format(LOG_DIR, now, logger_name))
        _tl_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
        _tl_fh.setFormatter(_tl_formatter)
        self._test_logger.addHandler(_tl_fh)

        # Create a console handler for the test logger.
        # This is the logger to the console
        _tl_console = logging.StreamHandler()
        self._test_logger.addHandler(_tl_console)

        # Create a notifier logger. This will notify via Telegram:
        # * start of a test suite
        # * Stop of a test suite
        # * Test suite and test case errors

        # Create an unlimited size queue and attach it to a queue handler
        que = queue.Queue(-1)
        queue_handler = QueueHandler(que)
        # Create a Telegram notification handler.  
        hdlr = NotificationHandler("telegram", defaults=defaults)

        # On the other side of the queue attach a queue listener
        self._listener = QueueListener(que, hdlr)
        queue_handler.setLevel(logging.ERROR)
        self._test_logger.addHandler(queue_handler)
        formatter = logging.Formatter("%(asctime)s - %(levelno)s - %(levelname)s - %(message)s")

        hdlr.setFormatter(formatter)

    def debug(self, msg):
        self._test_logger.debug(msg)

    def info(self, msg):
        self._test_logger.info(msg)

    def warning(self, msg):
        self._test_logger.warning(msg)

    def error(self, msg):
        self._test_logger.error(msg)

    def start_logger(self):
        self._listener.start()

    def stop_logger(self):
        self._listener.stop()

  2. 0

    You can use something like this:

    import logging
from logging import getLogger
from logging.handlers import SysLogHandler
from logging import StreamHandler,Formatter


DEFAULT_LOG_ADDRESS = ''      # host:port
DEFAULT_LOG_PERIOD  = '1000'  # positive integer
DEFAULT_LOG_LEVEL   = 'DEBUG' # DEBUG/INFO/WARNING/ERROR/CRITICAL


class Logger(name):
    def __init__(self):
        self.logger   = getLogger(name)
        self.debug    = self.logger.debug
        self.info     = self.logger.info
        self.warning  = self.logger.warning
        self.error    = self.logger.error
        self.critical = self.logger.critical
        self.period   = int(os.getenv('LOG_PERIOD',DEFAULT_LOG_PERIOD))
        level         = getattr(logging,os.getenv('LOG_LEVEL',DEFAULT_LOG_LEVEL).upper())
        address       = os.getenv('LOG_ADDRESS',DEFAULT_LOG_ADDRESS)
        if address:
            host,port = address.split(':')
            file_name = os.path.basename(sys.argv[0])
            log_message = Formatter(file_name+' %(message)s')
            syslogHandler = SysLogHandler(address=(host,int(port)))
            syslogHandler.setFormatter(log_message)
            self.logger.addHandler(syslogHandler)
        streamHandler = StreamHandler(sys.stdout)
        streamHandler.setLevel(level)
        self.logger.addHandler(streamHandler)
        self.logger.setLevel(level)
    def periodic(self,testCount,numOfTests,message):
        func = self.debug if testCount % self.period else self.info
        func('Test {} out of {}: {}'.format(testCount,numOfTests,message))

    Then you have the following functions at your disposal (to be used instead of print):

    • debug
    • info
    • warning
    • error
    • critical
    • periodic
    Login or Signup to reply.
  3. 0

    something I found elsewhere, and it works great!
    Auto-Start and Stop

    from queue import Queue
from atexit import register
from logging import LogRecord, getLogger
from logging.handlers import QueueHandler, QueueListener
from threading import RLock

_lock = RLock()


def _acquire_lock():
"""
Acquire the module-level lock for serializing access to shared data.
This should be released with _releaseLock().
"""
    if _lock:
        _lock.acquire()


def _release_lock():
"""
Release the module-level lock acquired by calling _acquireLock().
"""
    if _lock:
        _lock.release()


class LogQueueHandler(QueueHandler):
    def __init__(self, handlers, respect_handler_level=False, auto_run=True, queue=Queue(-1)):
        super().__init__(queue)
        self.queue = queue
        self._listener = QueueListener(self.queue, *handlers, respect_handler_level=respect_handler_level)
        if auto_run:
            self.start()
            register(self.stop)


    def start(self):
        self._listener.start()

    def stop(self):
        self._listener.stop()

    def emit(self, record: LogRecord) -> None:
        return super().emit(record)
    Login or Signup to reply.
  4. 0

    I’m not sure if this addresses your updated question directly, but for those coming here for the question title here is a way you can wrap the logging.Logger method and retain the full feature set of the logging module via wrapping and overwriting the root logger.

    After you have configured your logger and are able to get your logger using the following snippet we can start wrapping the logger.

    logger = logging.getLogger(LOGGER_NAME)

    The actual wrapping of the logger object can be done as follows. If the class that you’re object is instantiated from contains __slots__ this methodology will not work as expected. (see: https://stackoverflow.com/a/1445289/11770393)

    class LoggerWrapper(logging.Logger):

    def __init__(self, baseLogger):
        self.__class__ = type(baseLogger.__class__.__name__,
                             (self.__class__, baseLogger.__class__),
                             {})
        self.__dict__ = baseLogger.__dict__
        self.overwritten_property = ...

    def overwritten_method(self, *args):
        ...

    After we have created our wrapper class we can create an instance of this class doing the following.

    logger = LoggerWrapper(logging.getLogger(LOGGER_NAME))

    Now if you call logging methods directly against this object, they will perform based on the overwritten methods. However, we can take this a step further to overwrite the root logger in the logging module. By overwriting the root logger we can call logging.getLogger(LOGGER_NAME) in other files in our project. The following snippet will perform the overwrite.

    logging.root = LoggerWrapper(logging.getLogger(LOGGER_NAME))
logging.root.manager.loggerDict[LOGGER_NAME] = logging.root
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search