Python 日志模块

Python 日志模块

日志能把程序里出问题的信息记下来，我们靠它去查错误到底是怎么发生的。Python 的 logging 模块就是干这个的。它的工作流程是这样的：

记录器先产生日志信息，把它封装成一个 LogRecord 对象，接着检查这条日志的级别够不够处理门槛。不够就直接丢弃，够了就送进过滤器，通过过滤器后再交给处理器。

到了处理器这一层还会再查一次级别，符合的话又进一次过滤器，通过后才送到格式器做字符串格式化，最后写到指定设备上，比如标准输出、标准错误、文件、网络等等。

整个过程里有四个核心组件：

  1. 记录器：对外暴露应用代码直接调用的接口
  2. 处理器：把记录器产出的日志发到合适的目的地
  3. 过滤器：提供更精细的控制，决定哪些日志真正输出
  4. 格式器：规定最终输出里日志长什么样

1. 记录器

记录器对外暴露的就是应用代码直接调用的那些接口。用这些接口之前，得先搞清楚日志的级别。不同级别代表这条日志有多重要，输出时会受级别限制。比如我们把输出级别设成 ERROR，那么级别比 ERROR 低的日志就会被过滤掉，不会打出来。

数字越大，这条信息就越重要，忽略它给系统带来的安全隐患也越大。注意：设级别时，WARN 和 WARNING 是一回事，FATAL 和 CRITICAL 也是一回事。

另外注意：NOTSET 意指不设置，按照父 Logger 级别来过滤日志。

常用的几个直接调用接口是：

import logging

def test01():

    # 1. 创建日志对象
    logger = logging.getLogger()

    # 2. 默认日志级别
    print('日志级别:', logger.level)

    # 3. 设置日志级别
    logger.setLevel(logging.DEBUG)

    # 3. 打印不同日志
    logger.debug(msg='debug 日志')
    logger.info(msg='info 日志')
    logger.warning(msg='warning 日志')
    logger.error(msg='error 日志')
    logger.critical(msg='critical 日志')

if __name__ == '__main__':
    test01()

程序的输出结果：

日志级别: 30
warning 日志
error 日志
critical 日志

2. 处理器

上面这个例子里，我们明明调了 5 个级别的接口，结果只打出了 warning、error、critical 三条，另外两条没出来。

为什么没有输出？

先说明一点，我们已经把记录器默认的日志级别从 WARNING 改成了 DEBUG，记录器这一层其实已经不卡级别了。

这里就要引出 Handler（处理器）这个概念。处理器负责把日志送到指定目的地，所以很可能是目的地那头不认这个级别的日志。我们先看看常用的处理器有哪些：

FileHandler      用于将日志记录输出到文件中
StreamHandler    用于将日志记录输出到标准输出、或者标准错误中

logging 模块默认用的是 StreamHandler，而且这个默认 Handler 的级别被设成了 WARNING。我们只要把默认 Handler 的级别也改掉，就能打出所有级别的日志，如下：

# 未添加任何 Handler 时使用的默认 Handler
# _defaultLastResort = _StderrHandler(WARNING)
# lastResort = _defaultLastResort

def test02():

    # 1. 创建日志对象
    logger = logging.getLogger()

    # 2. 修改记录器日志级别
    logger.setLevel(logging.DEBUG)

    # 3. 修改默认处理器日志级别
    logging.lastResort.setLevel(logging.DEBUG)

    # 4. 打印不同日志
    logger.debug(msg='debug 日志')
    logger.info(msg='info 日志')
    logger.warning(msg='warning 日志')
    logger.error(msg='error 日志')
    logger.critical(msg='critical 日志')

_StderrHandler 是 StreamHandler 的子类，默认绑定了 WARNING 级别，源码如下：

class _StderrHandler(StreamHandler):
    """
    This class is like a StreamHandler using sys.stderr, but always uses
    whatever sys.stderr is currently set to rather than the value of
    sys.stderr at handler construction time.
    """
    def __init__(self, level=NOTSET):
        """
        Initialize the handler.
        """
        Handler.__init__(self, level)

    @property
    def stream(self):
        return sys.stderr


_defaultLastResort = _StderrHandler(WARNING)
# lastResort 使用的默认处理器
lastResort = _defaultLastResort

这里要分清标准错误和标准输出：标准输出是带缓冲区的，标准错误不带缓冲区。缓冲区分块缓冲和行缓冲两种，这里说的是行缓冲，文件读写用的则是块缓冲。也就是说，标准输出要碰到换行符或者缓冲区满了，才真正输出。

处理器的使用方法示例

构建 StreamHandler 时，可以指定 stream=stdout 把它绑到标准输出，指定 stream=stderr 把它绑到标准错误。

再说输出设备。日志可以打到标准输入输出，也可以写进文件，同理还能发到网络（这就要自己写 Handler 类了）。下面演示一下用法：

import logging
import sys

def test03():

    # 1. 获得 root 日志对象
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)

    # 2. 绑定多个处理器
    handler1 = logging.StreamHandler(sys.stderr)  # 关联标准错误
    handler1.setLevel(logging.INFO)
    logger.addHandler(handler1)

    handler2 = logging.StreamHandler(sys.stdout)  # 关联标准输出
    handler2.setLevel(logging.INFO)
    logger.addHandler(handler2)

    handler3 = logging.FileHandler(filename='demo.log')  # 关联文件输出
    handler3.setLevel(logging.INFO)
    logger.addHandler(handler3)

    # 4. 打印不同级别日志
    logger.debug(msg='debug 日志')
    logger.info(msg='info 日志')
    logger.warning(msg='warning 日志')
    logger.error(msg='error 日志')
    logger.critical(msg='critical 日志')

if __name__ == '__main__':
    test03()

程序执行结果：

info 日志
warning 日志
error 日志
critical 日志
info 日志
warning 日志
error 日志
critical 日志

还有一部分日志被写进了 demo.log 文件里。

层级结构下的处理器调用规则

记录器之间是有层级的。当一个子记录器自己没配处理器时，它会顺着层级往上找父记录器的处理器，父记录器有处理器就调用它。下面是 Logger 类里的源码实现：

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

3. 过滤器

Logger 和 Handler 都继承了 Filterer 类，两边都能挂过滤器。一条日志在传递过程中会被过滤两次。第一次是 Logger 处理这条记录时，按自己挂的过滤器列表过一遍，看 Logger 的 handle 方法：

def handle(self, record):
    """
    Call the handlers for the specified record.

    This method is used for unpickled records received from a socket, as
    well as those created locally. Logger-level filtering is applied.
    """
    if (not self.disabled) and self.filter(record):  # 执行过滤
        self.callHandlers(record)  # 调用处理器

第二次是日志传到 Handler 时，再按 Handler 自己挂的过滤器列表过一遍，然后才真正输出到设备，看 Handler 的 handle 方法：

def handle(self, record):
    """
    Conditionally emit the specified logging record.

    Emission depends on filters which may have been added to the handler.
    Wrap the actual emission of the record with acquisition/release of
    the I/O thread lock. Returns whether the filter passed the record for
    emission.
    """
    rv = self.filter(record)  # 执行过滤
    if rv:
        self.acquire()
        try:
            self.emit(record)  # 输出到设备
        finally:
            self.release()
    return rv

自定义过滤器示例：

import logging
import sys

class CustonLoggerFilter(object):

    def __init__(self):
        pass

    def filter(self, record):
        print('日志的记录器:', record.name)
        print('日志描述信息:', record.msg)
        print('日志发出文件:', record.filename)
        print('日志创建时间:', record.created)
        print('日志创建时间:', record.created)
        print('日志发出函数:', record.funcName)
        print('日志发出行号:', record.lineno)
        print('日志发出级别:', record.levelno)
        print('日志发出模块:', record.module)
        print('-' * 30)
        return True


def test04():

    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)

    # 记录器添加自定义过滤器
    filter = CustonLoggerFilter()
    logger.addFilter(filter)

    # 处理器添加自定义过滤器
    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(logging.DEBUG)
    handler.addFilter(filter)
    logger.addHandler(handler)

    logger.debug(msg='debug 日志')


if __name__ == '__main__':
    test04()

程序执行结果：

日志的记录器: root
日志描述信息: debug 日志
日志发出文件: 22-日志模块.py
日志创建时间: 1667240610.3610618
日志创建时间: 1667240610.3610618
日志发出函数: test04
日志发出行号: 107
日志发出级别: 10
日志发出模块: 22-日志模块
------------------------------
日志的记录器: root
日志描述信息: debug 日志
日志发出文件: 22-日志模块.py
日志创建时间: 1667240610.3610618
日志创建时间: 1667240610.3610618
日志发出函数: test04
日志发出行号: 107
日志发出级别: 10
日志发出模块: 22-日志模块
------------------------------
debug 日志

4. 格式器

格式器负责规定最终输出里日志的样式。它是挂在处理器上的，因为真正做输出的是处理器。

%(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

示例代码：

def test05():

    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)


    handler = logging.StreamHandler(sys.stderr)
    handler.setLevel(logging.DEBUG)
    # 构建输出格式器
    formatter = logging.Formatter('[%(levelname)s] "%(message)s" in %(pathname)s',
                                  datefmt='%Y-%m-%d %H:%M:%S')
    # 格式器设置到输出器
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    logger.debug(msg='日志信息')


if __name__ == '__main__':
    test05()

程序执行结果：

[DEBUG] "日志信息" in /Users/meng/Desktop/面试题/22-日志模块.py
©2026 老孟讲编程 京ICP备2021009384号anbei冀公网安备13050302001966号