python log with support for color printing and file size slicing and writing to mongodb

1. The custom ColorHandler and MongoHandler are used in the project, and the built-in RotatingFileHandler and the ConcurrentRotatingFileHandler of the three-party library are used.

Logs with different logger names are supported to be written to different files, and logs with different logger names are written to different mongodb collections. LogManager is easier to call, because all the internal methods inside use underscores. The underscores are protected and private methods that do not need to be called from the outside world, and do not need to understand him. When calling, pycharm will not automatically complete and prompt these underscores. It has nothing to do with the method, only exposes two methods that may need to be called, get_and_add_handlers and get_without_handlers, and pyrcharm can automatically complete these two methods.

2. The main ideas and models are:

After the logger adds a handler, every time you write logger.debug/info, it supports multiple logging methods, because the logger and handler are the relationship between the publisher and the subscriber, which is called the observer mode in the design mode. When logger.debug is performed, the logger will call its own _log method, the _log method calls the handle method, the handle calls the call handler method, the callhandler finds all subscribers, and calls the subscribers (the subscribers here are various addHandler is added to the handle method of the handler object of the list), and the handler method of the handler will call the emit method. So when writing a custom handler, you only need to inherit the logging.Handler class and rewrite the emit method. If there is a special need to pass in other parameters, in addition to the emit method, the init method needs to be rewritten.

So there are two main design patterns in it. The pattern used between logger and various handler objects is the observer pattern. The various handler classes and the basic Handler class use the template pattern (the template pattern is the main The method steps are written in the base class, and some specific steps are set as abstract methods in the template class, which must be rewritten by subclasses.)

# coding=utf8
"""
Log management, support log printing to console or write to file or mongodb
The usage is logger = LogManager('logger_name').get_and_add_handlers(log_level_int=1, is_add_stream_handler=True, log_path=None, log_filename=None, log_file_size=10, mongo_url=None, formatter_template=2)
Or logger = LogManager('logger_name').get_without_handlers(), this kind of log is not recorded immediately, and then all logs can be captured and recorded by get_and_add_handlers according to loggerame at a separate and unified main gate
"""
import os
import unittest
import time
import re
from collections import OrderedDict
import pymongo
import logging
from logging.handlers import RotatingFileHandler

if os.name == 'posix':
    from cloghandler import ConcurrentRotatingFileHandler

formatter_dict = {
    1: logging.Formatter('log time [%(asctime)s] - log name [%(name)s] - file [%(filename)s] - line [%(lineno)d] - log level [ %(levelname)s】 - log information【%(message)s】', "%Y-%m-%d %H:%M:%S"),
    2: logging.Formatter('%(asctime)s - %(name)s - %(filename)s - %(lineno)d - %(levelname)s - %(message)s', "%Y -%m-%d %H:%M:%S"),
}


class LogLevelException(Exception):
    def __init__(self, log_level):
        err = 'The set log level is {0}, the setting is wrong, please set it to a number in the range of 1 2 3 4 5'.format(log_level)
        Exception.__init__(self, err)


class MongoHandler(logging.Handler):
    """
    A log handler for mongodb, which supports logs to create different collections according to loggername and write them into mongodb
    """
    msg_pattern = re.compile('(\d + -\d + -\d + \d + :\d + :\d + ) - (\S*?) - (\ S*?) - (\d + ) - (\S*?) - ([\s\S]*)')

    def __init__(self, mongo_url, mongo_database='logs'):
        """
        :param mongo_url: mongo connection
        :param mongo_database: save the log ide database, the logs database is used by default
        """
        logging.Handler.__init__(self)
        mongo_client = pymongo. MongoClient(mongo_url)
        self.mongo_db = mongo_client.get_database(mongo_database)

    def emit(self, record):
        try:
            """The following uses the method of parsing the log template to extract the fields"""
            # msg = self. format(record)
            # logging. LogRecord
            # msg_match = self.msg_pattern.search(msg)
            # log_info_dict = {'time': msg_match.group(1),
            # 'name': msg_match. group(2),
            # 'file_name': msg_match. group(3),
            # 'line_no': msg_match. group(4),
            # 'log_level': msg_match. group(5),
            # 'detail_msg': msg_match.group(6),
            # }
            level_str = None
            if record.levelno == 10:
                level_str = 'DEBUG'
            elif record.levelno == 20:
                level_str = 'INFO'
            elif record.levelno == 30:
                level_str = 'WARNING'
            elif record.levelno == 40:
                level_str = 'ERROR'
            elif record.levelno == 50:
                level_str = 'CRITICAL'
            log_info_dict = OrderedDict()
            log_info_dict['time'] = time.strftime('%Y-%m-%d %H:%M:%S')
            log_info_dict['name'] = record.name
            log_info_dict['file_path'] = record.pathname
            log_info_dict['file_name'] = record.filename
            log_info_dict['func_name'] = record.funcName
            log_info_dict['line_no'] = record.lineno
            log_info_dict['log_level'] = level_str
            log_info_dict['detail_msg'] = record.msg
            col = self.mongo_db.get_collection(record.name)
            col.insert_one(log_info_dict)
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self. handleError(record)


class ColorHandler(logging. StreamHandler):
    """Colored logs, display different colors according to different levels of logs"""

    def emit(self, record):
        """
        0 40 black
        31 41 red
        32 42 green
        33 43 yellow
        34 44 blue
        35 45 fuchsia
        36 46 Cyan blue
        37 47 white
        :param record:
        :return:
        """
        try:
            #logging.LogRecord.levelno
            msg = self. format(record)

            if record.levelno == 10:
                print('\033[0;32m%s\033[0m' % msg) # green
            elif record.levelno == 20:
                print('\033[0;36m%s\033[0m' % msg) # blue
            elif record.levelno == 30:
                print('\033[0;34m%s\033[0m' % msg) # blue
            elif record.levelno == 40:
                print('\033[0;35m%s\033[0m' % msg) # fuchsia
            elif record.levelno == 50:
                print('\033[0;31m%s\033[0m' % msg) # blood red
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            self. handleError(record)


class LogManager(object):
    """
    A log class for creating and capturing logs, and supports printing logs to the console and writing to log files.
    """

    def __init__(self, logger_name=None):
        """
        :param logger_name: log name, when it is None, print all logs
        """
        self. logger = logging. getLogger(logger_name)
        self._logger_level = None
        self._is_add_stream_handler = None
        self._log_path = None
        self._log_filename = None
        self._log_file_size = None
        self._mongo_url = None
        self._formatter = None

    def get_logger_and_add_handlers(self, log_level_int=1, is_add_stream_handler=True, log_path=None, log_filename=None, log_file_size=10, mongo_url=None, formatter_template=2):
        """
       :param log_level_int: Log output level, set to 1 2 3 4 5, corresponding to output DEBUG, INFO, WARNING, ERROR, CRITICAL logs
       :param is_add_stream_handler: Whether to print the log to the console
       :param log_path: Set the folder path for storing logs
       :param log_filename: The name of the log, only when log_path and log_filename are not None, it will be written to the log file.
       :param log_file_size : log size, unit M, default 10M
       :param mongo_url : mongodb connection, no mongohandler will be added when it is None
       :param formatter_template : log template, 1 is the detailed template of formatter_dict, 2 is the brief template
       :type log_level_int :int
       :type is_add_stream_handler :bool
       :type log_path :str
       :type log_filename :str
       :type mongo_url :str
       :type log_file_size :int
       """
        self.__check_log_level(log_level_int)
        self._logger_level = self.__transform_logger_level(log_level_int)
        self._is_add_stream_handler = is_add_stream_handler
        self._log_path = log_path
        self._log_filename = log_filename
        self._log_file_size = log_file_size
        self._mongo_url = mongo_url
        self._formatter = formatter_dict[formatter_template]
        self.__set_logger_level()
        self.__add_handlers()
        return self. logger

    def get_logger_without_handlers(self):
        """Returns a logger without hanlers"""
        return self. logger

    def __set_logger_level(self):
        self.logger.setLevel(self._logger_level)

    @staticmethod
    def __check_log_level(log_level_int):
        if log_level_int not in [1, 2, 3, 4, 5]:
            raise LogLevelException(log_level_int)

    @staticmethod
    def __transform_logger_level(log_level_int):
        logger_level = None
        if log_level_int == 1:
            logger_level = logging. DEBUG
        elif log_level_int == 2:
            logger_level = logging. INFO
        elif log_level_int == 3:
            logger_level = logging. WARNING
        elif log_level_int == 4:
            logger_level = logging.ERROR
        elif log_level_int == 5:
            logger_level = logging. CRITICAL
        return logger_level

    def __add_handlers(self):
        if self._is_add_stream_handler:
            for h in self. logger. handlers:
                if isinstance(h, logging. StreamHandler):
                    break
            else:
                self.__add_stream_handler()
        if all([self._log_path, self._log_filename]):
            for h in self. logger. handlers:
                if os.name == 'nt':
                    if isinstance(h, RotatingFileHandler):
                        break
                if os.name == 'posix':
                    if isinstance(h, (RotatingFileHandler, ConcurrentRotatingFileHandler)):
                        break
            else:
                self.__add_file_handler()
        if self._mongo_url:
            for h in self. logger. handlers:
                if isinstance(h, MongoHandler):
                    break
            else:
                self.__add_mongo_handler()

    def __add_mongo_handler(self):
        """Write logs to mongodb"""
        mongo_handler = MongoHandler(self._mongo_url, 'logs')
        mongo_handler. setLevel(logging. DEBUG)
        mongo_handler.setFormatter(self._logger_level)
        self. logger. addHandler(mongo_handler)

    def __add_stream_handler(self):
        """
        Logs are displayed to the console
        """
        # stream_handler = logging. StreamHandler()
        stream_handler = ColorHandler() # Do not use streamhandler, use custom color log
        stream_handler.setLevel(self._logger_level)
        stream_handler.setFormatter(self._formatter)
        self. logger. addHandler(stream_handler)

    def __add_file_handler(self):
        """
        Logs are written to the log file
        """
        if not os.path.exists(self._log_path):
            os.makedirs(self._log_path)
        log_file = os.path.join(self._log_path, self._log_filename)
        os_name = os.name
        rotate_file_handler = None
        if os_name == 'nt':
            # Use this under windows, non-process safe
            rotate_file_handler = RotatingFileHandler(log_file, mode="a", maxBytes=self._log_file_size * 1024 * 1024, backupCount=10,
                                                      encoding="utf-8")
        if os_name == 'posix':
            # ConcurrentRotatingFileHandler can be used under linux, a process-safe log method
            rotate_file_handler = ConcurrentRotatingFileHandler(log_file, mode="a", maxBytes=self._log_file_size * 1024 * 1024,
                                                                backupCount=10, encoding="utf-8")
        rotate_file_handler.setLevel(self._logger_level)
        rotate_file_handler.setFormatter(self._formatter)
        self. logger. addHandler(rotate_file_handler)


class _Test(unittest. TestCase):
    @unittest.skip
    def test_repeat_add_handlers_(self):
        """Test repeated adding handlers"""
        LogManager('test').get_logger_and_add_handlers(1, log_path='../logs', log_filename='test.log')
        LogManager('test').get_logger_and_add_handlers(1, log_path='../logs', log_filename='test.log')
        LogManager('test').get_logger_and_add_handlers(1, log_path='../logs', log_filename='test.log')
        test_log = LogManager('test').get_logger_and_add_handlers(1, log_path='../logs', log_filename='test.log')
        print('The following sentence will not be printed four times and written to the log four times')
        time. sleep(1)
        test_log.debug('This sentence will not be printed four times and written to the log four times')

    @unittest.skip
    def test_get_logger_without_hanlders(self):
        """Test logs without handlers"""
        log = LogManager('test2').get_logger_without_handlers()
        print('The following sentence will not be printed')
        time. sleep(1)
        log.info('This sentence will not be printed')

    @unittest.skip
    def test_add_handlers(self):
        """In this way, you can write debug and info level logs at any specific place, and you only need to specify the level at the main gate to filter, which is very convenient"""
        LogManager('test3').get_logger_and_add_handlers(2)
        log1 = LogManager('test3').get_logger_without_handlers()
        print('The following sentence is info level and can be printed out')
        time. sleep(1)
        log1.info('This sentence is info level, it can be printed out')
        print('The following sentence is debug level and cannot be printed')
        time. sleep(1)
        log1.debug('This sentence is debug level and cannot be printed out')

    @unittest.skip
    def test_only_write_log_to_file(self):
        """Only write to log file"""
        log5 = LogManager('test5').get_logger_and_add_handlers(is_add_stream_handler=False, log_path='../logs', log_filename='test5.log')
        print('The following sentence is only written to the file')
        log5.debug('This sentence is only written to the file')

    def test_color_and_mongo_hanlder(self):
        """Test color log and log write to mongodb"""
        from app import config
        logger = LogManager('helloMongo').get_logger_and_add_handlers(mongo_url=config.connect_url)
        logger.debug('a debug level log')
        logger.info('an info level log')
        logger.warning('a warning level log')
        logger.error('a log of error level')
        logger.critical('a critical level log')


if __name__ == "__main__":
    unittest. main()
The effect of test_color_and_mongo_hanlder is as follows.<br>3. Because a custom ColorHandler is added to the logger, the logs recorded by the console are colored. 

4. In mongodb

Use the name attribute of the logger as the name of the collection. Different loggernames create different collections. The main saved fields are the log name, what level of logs and log details are printed in which line of which function/method in which file.

The knowledge points of the article match the official knowledge files, and you can further learn relevant knowledgePython introductory skill treeHomepageOverview 258598 people are studying systematically