Python開發——logging模塊詳解

logging模塊詳解

logging簡介

logging是python的內置庫，主要用於進行格式化內容輸出，可將格式化內容輸出到文件，也可輸出到屏幕。咱們在開發過程當中經常使用print函數來進行調試，可是實際應用部署時咱們要將日誌的信息要輸出到文件中，方便後續查找以及備份。在咱們使用日誌管理時，咱們還能夠將日誌格式化成json對象轉存到ELK中方便圖形化查看及管理。前面說的這些，咱們均可以經過logging所包含的功能以及提供擴展的方式來完成。

logging工做流程

這是從網上查到一個關於python logging模塊的工做流程圖，很是遺憾沒有找到出處。

能夠看到圖中，幾個關鍵的類

Logger

用於記錄日誌的對象。

經過流程圖能夠看到

1. 判斷是否enabled，實質就是看記錄的level（logger.info，logger.debug等）和當前logger對象設置的level是否知足（好比logger設置的lever是Info，記錄時使用的logger.debug，那麼就會不知足，因此不會記錄日誌）
2. 查看logger的過濾器是否知足。filter經過以後，交給logger的handler來記錄日誌，一個logger是能夠設置多個handler。
3. 交給Handlers實際記錄日誌

注：在整個應用中能夠有多個logger，使用logging.getLogger時經過指定name來獲取對象，實際logging中還存在一個Manager類，由Manager來進行多logger的單例模式管理。

Handler

用於記錄日誌到具體的文件或者輸出流或其餘的管道。

1. 查看記錄日誌是否知足過濾器
2. 知足過濾器，按照設置的Formatter生成字符串
3. 將內容寫入到具體的文件或者輸出流

不一樣的Handler可能有不一樣的處理，可是底層原理仍是作這三件事情。

Filter

用於過濾用戶記錄日誌時，是否知足記錄標準

Formatter

用於格式化，輸出的字符串格式

這是原生自帶的格式

%(name)s            Name of the logger (logging channel)
    %(levelno)s         Numeric logging level for the message (DEBUG, INFO,
                        WARNING, ERROR, CRITICAL)
    %(levelname)s       Text logging level for the message ("DEBUG", "INFO",
                        "WARNING", "ERROR", "CRITICAL")
    %(pathname)s        Full pathname of the source file where the logging
                        call was issued (if available)
    %(filename)s        Filename portion of pathname
    %(module)s          Module (name portion of filename)
    %(lineno)d          Source line number where the logging call was issued
                        (if available)
    %(funcName)s        Function name
    %(created)f         Time when the LogRecord was created (time.time()
                        return value)
    %(asctime)s         Textual time when the LogRecord was created
    %(msecs)d           Millisecond portion of the creation time
    %(relativeCreated)d Time in milliseconds when the LogRecord was created,
                        relative to the time the logging module was loaded
                        (typically at application startup time)
    %(thread)d          Thread ID (if available)
    %(threadName)s      Thread name (if available)
    %(process)d         Process ID (if available)
    %(message)s         The result of record.getMessage(), computed just as
                        the record is emitted
複製代碼

LogRecord

咱們每一次的 logger.info logger.debug logger.error等實際上都是進行一次LogRecord的處理

包含一些基本的輸出日誌內容，以及內容中參數，還有附帶的函數名稱，行數等信息

logging源碼閱讀

這是我這邊基於字典的一個logging配置

import logging
import logging.config
import os

basedir = os.path.abspath(os.path.dirname(__file__))
log_path = os.path.join(basedir, '..', 'logs')
service_name = "test_log"
if not os.path.isdir(log_path):
    os.mkdir(log_path)
    
class InfoFilter(logging.Filter):
    def filter(self, record):
        """ 只篩選出INFO級別的日誌 """
        if logging.INFO <= record.levelno < logging.ERROR:
            return super().filter(record)
        else:
            return 0


class ErrorFilter(logging.Filter):
    def filter(self, record):
        """ 只篩選出ERROR級別的日誌 """
        if logging.ERROR <= record.levelno < logging.CRITICAL:
            return super().filter(record)
        else:
            return 0

LOG_CONFIG_DICT = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'simple': {
            'class': 'logging.Formatter',
            'format': '%(asctime)s %(levelname)s %(name)s %(filename)s %(module)s %(funcName)s '
                      '%(lineno)d %(thread)d %(threadName)s %(process)d %(processName)s %(message)s'
        },
        # json模式, 方便ELK收集處理
        'json': {
            'class': 'logging.Formatter',
            'format': '{"time:":"%(asctime)s","level":"%(levelname)s","logger_name":"%(name)s",'
                      '"file_name":"%(filename)s","module":"%(module)s","func_name":"%(funcName)s",'
                      '"line_number":"%(lineno)d","thread_id":"%(thread)d","thread_name":"%(threadName)s",'
                      '"process_id":"%(process)d","process_name":"%(processName)s","message":"%(message)s"}'}
    },
    # 過濾器
    'filters': {
        'info_filter': {
            '()': InfoFilter
        },
        'error_filter': {
            '()': ErrorFilter
        }
    },
    # 處理器
    'handlers': {
        # 控制檯輸出
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'simple'
        },
        # info文件輸出
        'info_file': {
            'level': 'INFO',
            'formatter': 'json',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'filename': '{0}/{1}_info.log'.format(log_path, service_name),
            'when': "d",
            'interval': 1,
            'encoding': 'utf8',
            'backupCount': 30,
            'filters': ['info_filter']
        },
        # error文件輸出
        'error_file': {
            'level': 'ERROR',
            'formatter': 'json',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'filename': '{0}/{1}_error.log'.format(log_path, service_name),
            'when': "d",
            'interval': 1,
            'encoding': 'utf8',
            'backupCount': 30,
            'filters': ['error_filter']
        }
    },
    # 記錄器
    'loggers': {
        'full_logger': {
            'handlers': ['console', 'info_file', 'error_file'],
            'level': 'INFO'
        },
        'only_console_logger': {
            'handlers': ['console'],
            'level': 'INFO'
        },
        'only_file_logger': {
            'handlers': ['info_file', 'error_file']
        }
    }
}

logging.config.dictConfig(LOG_CONFIG_DICT)
複製代碼

下面咱們就基於config第一次配置時整個logging的工做原理。結合代碼進行分析logging的工做之路。

個人python版本是Python3.6

從logging.config開始

dictConfigClass = DictConfigurator

def dictConfig(config):
    """Configure logging using a dictionary."""
    dictConfigClass(config).configure()
複製代碼

實質上是實例化logging.config.DictConfigurator類的對象，而後執行其configure方法。

class DictConfigurator(BaseConfigurator):
    """ Configure logging using a dictionary-like object to describe the configuration. """

    def configure(self):
        """Do the configuration."""

        config = self.config
        if 'version' not in config:
            raise ValueError("dictionary doesn't specify a version")
        if config['version'] != 1:
            raise ValueError("Unsupported version: %s" % config['version'])
        incremental = config.pop('incremental', False)
        EMPTY_DICT = {}
        logging._acquireLock()
        try:
            if incremental:
                handlers = config.get('handlers', EMPTY_DICT)
                for name in handlers:
                    if name not in logging._handlers:
                        raise ValueError('No handler found with '
                                         'name %r'  % name)
                    else:
                        try:
                            handler = logging._handlers[name]
                            handler_config = handlers[name]
                            level = handler_config.get('level', None)
                            if level:
                                handler.setLevel(logging._checkLevel(level))
                        except Exception as e:
                            raise ValueError('Unable to configure handler '
                                             '%r: %s' % (name, e))
                loggers = config.get('loggers', EMPTY_DICT)
                for name in loggers:
                    try:
                        self.configure_logger(name, loggers[name], True)
                    except Exception as e:
                        raise ValueError('Unable to configure logger '
                                         '%r: %s' % (name, e))
                root = config.get('root', None)
                if root:
                    try:
                        self.configure_root(root, True)
                    except Exception as e:
                        raise ValueError('Unable to configure root '
                                         'logger: %s' % e)
            else:
                disable_existing = config.pop('disable_existing_loggers', True)

                logging._handlers.clear()
                del logging._handlerList[:]

                # Do formatters first - they don't refer to anything else
                formatters = config.get('formatters', EMPTY_DICT)
                for name in formatters:
                    try:
                        formatters[name] = self.configure_formatter(
                                                            formatters[name])
                    except Exception as e:
                        raise ValueError('Unable to configure '
                                         'formatter %r: %s' % (name, e))
                # Next, do filters - they don't refer to anything else, either
                filters = config.get('filters', EMPTY_DICT)
                for name in filters:
                    try:
                        filters[name] = self.configure_filter(filters[name])
                    except Exception as e:
                        raise ValueError('Unable to configure '
                                         'filter %r: %s' % (name, e))
                handlers = config.get('handlers', EMPTY_DICT)
                deferred = []
                for name in sorted(handlers):
                    try:
                        handler = self.configure_handler(handlers[name])
                        handler.name = name
                        handlers[name] = handler
                    except Exception as e:
                        if 'target not configured yet' in str(e):
                            deferred.append(name)
                        else:
                            raise ValueError('Unable to configure handler '
                                             '%r: %s' % (name, e))

                # Now do any that were deferred
                for name in deferred:
                    try:
                        handler = self.configure_handler(handlers[name])
                        handler.name = name
                        handlers[name] = handler
                    except Exception as e:
                        raise ValueError('Unable to configure handler '
                                         '%r: %s' % (name, e))

                root = logging.root
                existing = list(root.manager.loggerDict.keys())

                existing.sort()

                child_loggers = []
                #now set up the new ones...
                loggers = config.get('loggers', EMPTY_DICT)
                for name in loggers:
                    if name in existing:
                        i = existing.index(name) + 1 # look after name
                        prefixed = name + "."
                        pflen = len(prefixed)
                        num_existing = len(existing)
                        while i < num_existing:
                            if existing[i][:pflen] == prefixed:
                                child_loggers.append(existing[i])
                            i += 1
                        existing.remove(name)
                    try:
                        self.configure_logger(name, loggers[name])
                    except Exception as e:
                        raise ValueError('Unable to configure logger '
                                         '%r: %s' % (name, e))

                _handle_existing_loggers(existing, child_loggers,
                                         disable_existing)

                # And finally, do the root logger
                root = config.get('root', None)
                if root:
                    try:
                        self.configure_root(root)
                    except Exception as e:
                        raise ValueError('Unable to configure root '
                                         'logger: %s' % e)
        finally:
            logging._releaseLock()
複製代碼

咱們對incremental不進行設置，即便用全量的方式進行配置，對於全部的handler重置並處理

看下來基本上分爲4步來走。

1. 配置Formatter
2. 配置Filter
3. 配置Handler
4. 配置Logger

Formatter配置

源碼部分

def configure_formatter(self, config):
        """Configure a formatter from a dictionary."""
        if '()' in config:
            factory = config['()'] # for use in exception handler
            try:
                result = self.configure_custom(config)
            except TypeError as te:
                if "'format'" not in str(te):
                    raise
                config['fmt'] = config.pop('format')
                config['()'] = factory
                result = self.configure_custom(config)
        else:
            fmt = config.get('format', None)
            dfmt = config.get('datefmt', None)
            style = config.get('style', '%')
            cname = config.get('class', None)
            if not cname:
                c = logging.Formatter
            else:
                c = _resolve(cname)
            result = c(fmt, dfmt, style)
        return result
複製代碼

能夠使用本身的Formatter，默認使用logging.Formatter

Filter配置

def configure_filter(self, config):
        """Configure a filter from a dictionary."""
        if '()' in config:
            result = self.configure_custom(config)
        else:
            name = config.get('name', '')
            result = logging.Filter(name)
        return result
複製代碼

設置filter,能夠參考我上面配置的字典，能夠本身定義Filter也能夠使用系統內置Filter

Handler配置

def configure_handler(self, config):
        """Configure a handler from a dictionary."""
        config_copy = dict(config)  # for restoring in case of error
        formatter = config.pop('formatter', None)
        if formatter:
            try:
                formatter = self.config['formatters'][formatter]
            except Exception as e:
                raise ValueError('Unable to set formatter '
                                 '%r: %s' % (formatter, e))
        level = config.pop('level', None)
        filters = config.pop('filters', None)
        if '()' in config:
            c = config.pop('()')
            if not callable(c):
                c = self.resolve(c)
            factory = c
        else:
            cname = config.pop('class')
            klass = self.resolve(cname)
            #Special case for handler which refers to another handler
            if issubclass(klass, logging.handlers.MemoryHandler) and\
                'target' in config:
                try:
                    th = self.config['handlers'][config['target']]
                    if not isinstance(th, logging.Handler):
                        config.update(config_copy)  # restore for deferred cfg
                        raise TypeError('target not configured yet')
                    config['target'] = th
                except Exception as e:
                    raise ValueError('Unable to set target handler '
                                     '%r: %s' % (config['target'], e))
            elif issubclass(klass, logging.handlers.SMTPHandler) and\
                'mailhost' in config:
                config['mailhost'] = self.as_tuple(config['mailhost'])
            elif issubclass(klass, logging.handlers.SysLogHandler) and\
                'address' in config:
                config['address'] = self.as_tuple(config['address'])
            factory = klass
        props = config.pop('.', None)
        kwargs = dict([(k, config[k]) for k in config if valid_ident(k)])
        try:
            result = factory(**kwargs)
        except TypeError as te:
            if "'stream'" not in str(te):
                raise
            kwargs['strm'] = kwargs.pop('stream')
            result = factory(**kwargs)
        if formatter:
            result.setFormatter(formatter)
        if level is not None:
            result.setLevel(logging._checkLevel(level))
        if filters:
            self.add_filters(result, filters)
        if props:
            for name, value in props.items():
                setattr(result, name, value)
        return result
複製代碼

能夠看到，前面順序先配置formatter和filter是由於這裏須要引用的到

值得注意的是，咱們初始化時傳遞的一個字典，在整個配置過程當中，字典裏面的值會隨着咱們每次的配置變化而變化，因此咱們在每一個元素配置以後，在使用上一個字典元素時，就是配置完成以後的元素，爲了方便理解，將配置filter以前和配置filter以後，config中的filter變化列出來

config == > before filters {'info_filter': {'()': <class 'utils.log.InfoFilter'>}, 'error_filter': {'()': <class 'utils.log.ErrorFilter'>}} 
config == > after filters {'info_filter': <utils.log.InfoFilter object at 0x00000000028EB208>, 'error_filter': <utils.log.ErrorFilter object at 0x00000000028EB240>}
複製代碼

配置前是咱們配置文件中的內容，配置完成以後filter已是一組對象了，因此在配置handler時咱們就能夠直接使用對象add_filter了。

Logger配置

def common_logger_config(self, logger, config, incremental=False):
        level = config.get('level', None)
        if level is not None:
            logger.setLevel(logging._checkLevel(level))
        if not incremental:
            #Remove any existing handlers
            for h in logger.handlers[:]:
                logger.removeHandler(h)
            handlers = config.get('handlers', None)
            if handlers:
                self.add_handlers(logger, handlers)
            filters = config.get('filters', None)
            if filters:
                self.add_filters(logger, filters)

    def configure_logger(self, name, config, incremental=False):
        """Configure a non-root logger from a dictionary."""
        logger = logging.getLogger(name)
        self.common_logger_config(logger, config, incremental)
        propagate = config.get('propagate', None)
        if propagate is not None:
            logger.propagate = propagate
複製代碼

這就比較容易理解了，將咱們上面配置過的filters和handlers添加到咱們的logger中。

這裏須要注意的一點是logger = logging.getLogger(name)，看下logger.getLogger源碼

root = RootLogger(WARNING)
Logger.root = root
Logger.manager = Manager(Logger.root)
def getLogger(name=None):
    if name:
        return Logger.manager.getLogger(name)
    else:
        return root
class Manager(object):
    def __init__(self, rootnode):
        self.root = rootnode
        self.disable = 0
        self.emittedNoHandlerWarning = False
        self.loggerDict = {}
        self.loggerClass = None
        self.logRecordFactory = None

    def getLogger(self, name):
        rv = None
        if not isinstance(name, str):
            raise TypeError('A logger name must be a string')
        _acquireLock()
        try:
            if name in self.loggerDict:
                rv = self.loggerDict[name]
                if isinstance(rv, PlaceHolder):
                    ph = rv
                    rv = (self.loggerClass or _loggerClass)(name)
                    rv.manager = self
                    self.loggerDict[name] = rv
                    self._fixupChildren(ph, rv)
                    self._fixupParents(rv)
            else:
                rv = (self.loggerClass or _loggerClass)(name)
                rv.manager = self
                self.loggerDict[name] = rv
                self._fixupParents(rv)
        finally:
            _releaseLock()
        return rv
複製代碼

能夠看到，logging使用Mangaer進行logger的單例管理

截止到這裏，基本上咱們使用前的準備，logging都替咱們準備好了，下面就是咱們的使用

獲取logger並記錄日誌

class Logger(Filterer):
	
	def info(self, msg, *args, **kwargs):
        if self.isEnabledFor(INFO):
            self._log(INFO, msg, args, **kwargs)
    
    def error(self, msg, *args, **kwargs):
        if self.isEnabledFor(ERROR):
            self._log(ERROR, msg, args, **kwargs)
    
    def isEnabledFor(self, level):
        if self.manager.disable >= level:
            return False
        return level >= self.getEffectiveLevel()
	
    def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False):
        sinfo = None
        if _srcfile:
            try:
                fn, lno, func, sinfo = self.findCaller(stack_info)
            except ValueError: # pragma: no cover
                fn, lno, func = "(unknown file)", 0, "(unknown function)"
        else: # pragma: no cover
            fn, lno, func = "(unknown file)", 0, "(unknown function)"
        if exc_info:
            if isinstance(exc_info, BaseException):
                exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
            elif not isinstance(exc_info, tuple):
                exc_info = sys.exc_info()
        record = self.makeRecord(self.name, level, fn, lno, msg, args,
                                 exc_info, func, extra, sinfo)
        self.handle(record)

    def handle(self, record):
        if (not self.disabled) and self.filter(record):
            self.callHandlers(record)
            
    def callHandlers(self, record):
        c = self
        found = 0
        while c:
            for hdlr in c.handlers:
                found = found + 1
                if record.levelno >= hdlr.level:
                    hdlr.handle(record)
            if not c.propagate:
                c = None    #break out
            else:
                c = c.parent
        if (found == 0):
            if lastResort:
                if record.levelno >= lastResort.level:
                    lastResort.handle(record)
            elif raiseExceptions and not self.manager.emittedNoHandlerWarning:
                sys.stderr.write("No handlers could be found for logger"
                                 " \"%s\"\n" % self.name)
                self.manager.emittedNoHandlerWarning = True
class StreamHandler(Handler):

    terminator = '\n'

    def __init__(self, stream=None):
        """ Initialize the handler. If stream is not specified, sys.stderr is used. """
        Handler.__init__(self)
        if stream is None:
            stream = sys.stderr
        self.stream = stream

    def flush(self):
        self.acquire()
        try:
            if self.stream and hasattr(self.stream, "flush"):
                self.stream.flush()
        finally:
            self.release()

    def emit(self, record):
        try:
            msg = self.format(record)
            stream = self.stream
            stream.write(msg)
            stream.write(self.terminator)
            self.flush()
        except Exception:
            self.handleError(record)

    def __repr__(self):
        level = getLevelName(self.level)
        name = getattr(self.stream, 'name', '')
        if name:
            name += ' '
        return '<%s %s(%s)>' % (self.__class__.__name__, name, level)
複製代碼

根據代碼能夠看到符合咱們流程圖看到的流程，咱們細化一下就是

1. 查看記錄日誌是否知足過濾器，而後準備logrecord中的信息並生成logrecord
2. 將logrecord交給全部handlers處理
3. 在handler中肯定是否知足handler過濾器，知足的話按照配置的Formatter生成字符串
4. 將內容寫入到具體的文件或者輸出流

logger = logging.getLogger("full_logger")

logger.info("111")
# 根據咱們的配置 這裏會輸出到流中，以及記錄到test_log_info.log文件中

logger.error("error")
# 根據咱們的配置 這裏會輸出到流中，以及記錄到test_log_error.info文件中

複製代碼

有興趣的同窗能夠繼續試試其餘好玩的處理

也能夠關注個人公衆號共同窗習。