日志模块介绍：掌握 Python 中的诊断日志记录

简介

本节简要介绍日志记录模块。

日志记录模块

logging 模块是一个用于记录诊断信息的标准库模块。它也是一个非常庞大的模块，具有许多复杂的功能。我们将展示一个简单的示例来说明它的用途。

重温异常

在练习中，我们编写了一个名为 parse() 的函数，大致如下：

## fileparse.py
def parse(f, types=None, names=None, delimiter=None):
    records = []
    for line in f:
        line = line.strip()
        if not line: continue
        try:
            records.append(split(line,types,names,delimiter))
        except ValueError as e:
            print("无法解析 :", line)
            print("原因 :", e)
    return records

关注 try-except 语句。在 except 块中你应该做什么？

你应该打印一条警告消息吗？

try:
    records.append(split(line,types,names,delimiter))
except ValueError as e:
    print("无法解析 :", line)
    print("原因 :", e)

还是默默地忽略它？

try:
    records.append(split(line,types,names,delimiter))
except ValueError as e:
    pass

这两种解决方案都不尽如人意，因为你通常希望两种行为（用户可选择）都具备。

使用日志记录

logging 模块可以解决这个问题。

## fileparse.py
import logging
log = logging.getLogger(__name__)

def parse(f,types=None,names=None,delimiter=None):
 ...
    try:
        records.append(split(line,types,names,delimiter))
    except ValueError as e:
        log.warning("无法解析 : %s", line)
        log.debug("原因 : %s", e)

代码被修改为使用一个特殊的 Logger 对象（通过 logging.getLogger(__name__) 创建）来发出警告消息。

日志记录基础

创建一个日志记录器对象。

log = logging.getLogger(name)   ## name 是一个字符串

发出日志消息。

log.critical(message [, args])
log.error(message [, args])
log.warning(message [, args])
log.info(message [, args])
log.debug(message [, args])

每个方法代表不同的严重级别。

所有这些方法都会创建一个格式化的日志消息。args 与 % 运算符一起用于创建消息。

logmsg = message % args ## 写入日志

日志记录配置

日志记录行为是单独配置的。

## main.py

...

if __name__ == '__main__':
    import logging
    logging.basicConfig(
        filename  = 'app.log',      ## 日志输出文件
        level     = logging.INFO,   ## 输出级别
    )

通常，这是在程序启动时进行的一次性配置。该配置与进行日志记录调用的代码是分开的。

注释

日志记录具有高度的可配置性。你可以调整它的各个方面：输出文件、级别、消息格式等等。然而，使用日志记录的代码无需担心这些。

练习8.2：向模块添加日志记录

在fileparse.py中，有一些与错误输入导致的异常相关的错误处理。它看起来像这样：

## fileparse.py
import csv

def parse_csv(lines, select=None, types=None, has_headers=True, delimiter=',', silence_errors=False):
    '''
    将CSV文件解析为具有类型转换的记录列表。
    '''
    if select and not has_headers:
        raise RuntimeError('select requires column headers')

    rows = csv.reader(lines, delimiter=delimiter)

    ## 读取文件头（如果有）
    headers = next(rows) if has_headers else []

    ## 如果选择了特定列，创建用于过滤的索引并设置输出列
    if select:
        indices = [ headers.index(colname) for colname in select ]
        headers = select

    records = []
    for rowno, row in enumerate(rows, 1):
        if not row:     ## 跳过无数据的行
            continue

        ## 如果选择了特定列索引，挑选出这些列
        if select:
            row = [ row[index] for index in indices]

        ## 对行应用类型转换
        if types:
            try:
                row = [func(val) for func, val in zip(types, row)]
            except ValueError as e:
                if not silence_errors:
                    print(f"第{rowno}行：无法转换 {row}")
                    print(f"第{rowno}行：原因 {e}")
                continue

        ## 创建字典或元组
        if headers:
            record = dict(zip(headers, row))
        else:
            record = tuple(row)
        records.append(record)

    return records

注意那些发出诊断消息的打印语句。用日志记录操作替换这些打印相对简单。将代码修改如下：

## fileparse.py
import csv
import logging
log = logging.getLogger(__name__)

def parse_csv(lines, select=None, types=None, has_headers=True, delimiter=',', silence_errors=False):
    '''
    将CSV文件解析为具有类型转换的记录列表。
    '''
    if select and not has_headers:
        raise RuntimeError('select requires column headers')

    rows = csv.reader(lines, delimiter=delimiter)

    ## 读取文件头（如果有）
    headers = next(rows) if has_headers else []

    ## 如果选择了特定列，创建用于过滤的索引并设置输出列
    if select:
        indices = [ headers.index(colname) for colname in select ]
        headers = select

    records = []
    for rowno, row in enumerate(rows, 1):
        if not row:     ## 跳过无数据的行
            continue

        ## 如果选择了特定列索引，挑选出这些列
        if select:
            row = [ row[index] for index in indices]

        ## 对行应用类型转换
        if types:
            try:
                row = [func(val) for func, val in zip(types, row)]
            except ValueError as e:
                if not silence_errors:
                    log.warning("第 %d 行：无法转换 %s", rowno, row)
                    log.debug("第 %d 行：原因 %s", rowno, e)
                continue

        ## 创建字典或元组
        if headers:
            record = dict(zip(headers, row))
        else:
            record = tuple(row)
        records.append(record)

    return records

现在你已经做了这些更改，尝试在错误数据上使用你的一些代码。

>>> import report
>>> a = report.read_portfolio('missing.csv')
第4行：错误的行：['MSFT', '', '51.23']
第7行：错误的行：['IBM', '', '70.44']
>>>

如果你不做任何操作，你只会得到WARNING级别及以上的日志消息。输出看起来会像简单的打印语句。然而，如果你配置日志记录模块，你会得到关于日志级别、模块等的更多信息。按以下步骤操作来查看：

>>> import logging
>>> logging.basicConfig()
>>> a = report.read_portfolio('missing.csv')
WARNING:fileparse:第4行：无法转换 ['MSFT', '', '51.23']
WARNING:fileparse:第7行：无法转换 ['IBM', '', '70.44']
>>>

你会注意到看不到log.debug()操作的输出。输入以下内容来更改级别。

>>> logging.getLogger('fileparse').setLevel(logging.DEBUG)
>>> a = report.read_portfolio('missing.csv')
WARNING:fileparse:第4行：无法转换 ['MSFT', '', '51.23']
DEBUG:fileparse:第4行：原因：invalid literal for int() with base 10: ''
WARNING:fileparse:第7行：无法转换 ['IBM', '', '70.44']
DEBUG:fileparse:第7行：原因：invalid literal for int() with base 10: ''
>>>

关闭所有日志消息，只保留最关键的：

>>> logging.getLogger('fileparse').setLevel(logging.CRITICAL)
>>> a = report.read_portfolio('missing.csv')
>>>

练习8.3：向程序添加日志记录

要向应用程序添加日志记录，你需要有某种机制在主模块中初始化日志记录模块。一种方法是包含一些如下所示的设置代码：

## 此文件设置日志记录模块的基本配置。
## 根据需要在此处更改设置以调整日志输出。
import logging
logging.basicConfig(
    filename = 'app.log',            ## 日志文件的名称（省略则使用stderr）
    filemode = 'w',                  ## 文件模式（使用'a'进行追加）
    level    = logging.WARNING,      ## 日志记录级别（DEBUG、INFO、WARNING、ERROR或CRITICAL）
)

同样，你需要将此代码放在程序启动步骤中的某个位置。例如，在你的report.py程序中，你会把它放在哪里呢？

✨ 查看解决方案并练习

总结

恭喜你！你已经完成了日志记录实验。你可以在LabEx中练习更多实验来提升你的技能。