目 录CONTENT

文章目录

优雅的日志记录方案之loguru

shibuyu
2022-06-14 / 0 评论 / 1 点赞 / 469 阅读 / 4,593 字 / 正在检测是否收录...
温馨提示:
本文最后更新于 2022-11-08,若内容或图片失效,请留言反馈。部分素材来自网络,若不小心影响到您的利益,请联系我们删除。

介绍

在 Python 中,通常我们使用自带的 logging 模块来记录日志,但需要配置 Handler、Formatter 等,步骤很繁琐。现在我们有了新的选择—loguru

安装

开箱即用,无需样板
首先,这个库的安装方式很简单,就用基本的 pip 安装即可,Python 3 版本的安装如下:

pip3 install loguru

安装完毕之后,我们就可以在项目里使用这个 loguru 库了。

from loguru import logger

logger.debug("Welcome to my blog--不予の測試筆記!")
logger.info('Welcome to my blog--不予の測試筆記!')
logger.warning('Welcome to my blog--不予の測試筆記!')
logger.error('Welcome to my blog--不予の測試筆記!')

输出如下

C:\Users\12446\AppData\Local\Programs\Python\Python39\python.exe D:/Lemon/作业/2.py
2022-06-14 20:11:26.536 | DEBUG    | __main__:<module>:9 - Welcome to my blog--不予の測試筆記!
2022-06-14 20:11:26.537 | INFO     | __main__:<module>:10 - Welcome to my blog--不予の測試筆記!
2022-06-14 20:11:26.537 | WARNING  | __main__:<module>:11 - Welcome to my blog--不予の測試筆記!
2022-06-14 20:11:26.537 | ERROR    | __main__:<module>:12 - Welcome to my blog--不予の測試筆記!

以下是具有默认严重性值的标准级别,每个级别都与同名的日志记录方法相关联:

级别名称 严重性值 记录器方法
TRACE 5 logger.trace()
DEBUG 10 logger.debug()
INFO 20 logger.info()
SUCCESS 25 logger.success()
WARNING 30 logger.warning()
ERROR 40 logger.error()
CRITICAL 50 logger.critical()

在 IDE 或终端运行时会发现,loguru 在输出的不同级别信息时,带上了不同的颜色,使得结果更加直观,其中也包含了时间、级别、模块名、行号以及日志信息。

logger 默认采用 sys.stderr 标准输出将日志输出到控制台中,并没有输出到其他的地方,如果想要输出到其他的位置,比如存为文件,我们只需要使用一行代码声明即可。 例如将结果同时输出到一个 runtime.log 文件里面,可以这么写:

from loguru import logger

logger.add('runtime.log')
logger.debug("Welcome to my blog--不予の測試筆記!")
logger.info('Welcome to my blog--不予の測試筆記!')
logger.warning('Welcome to my blog--不予の測試筆記!')
logger.error('Welcome to my blog--不予の測試筆記!')

不需要再声明一个 FileHandler 了,就一行 add 语句搞定,运行之后会发现目录下 runtime.log 里面同样出现了刚刚控制台输出的 DEBUG 信息

详细使用

loguru 对输出到文件的配置有非常强大的支持,比如支持输出到多个文件,分级别分别输出,过大创建新文件,过久自动删除等等。 下面我们分别看看这些怎样来实现,这里基本上就是 add 方法的使用介绍。因为这个 add 方法就相当于给 logger 添加了一个 Handler,它给我们暴露了许多参数来实现 Handler 的配置,下面我们来详细介绍下。 首先看看它的方法定义吧:

    def add(
        self,
        sink: Union[TextIO, Writable, Callable[[Message], None], Handler],
        *,
        level: Union[str, int] = ...,
        format: Union[str, FormatFunction] = ...,
        filter: Optional[Union[str, FilterFunction, FilterDict]] = ...,
        colorize: Optional[bool] = ...,
        serialize: bool = ...,
        backtrace: bool = ...,
        diagnose: bool = ...,
        enqueue: bool = ...,
        catch: bool = ...
    ) -> int: ...

看看它的源代码,它支持这么多的参数,如 level、format、filter、color 等等,另外我们还注意到它有个非常重要的参数 sink,我们看看官方文档

  • sink 可以传入一个 file 对象,例如 sys.stderr 或者 open('file.log', 'w') 都可以。

  • sink 可以直接传入一个 str 字符串或者 pathlib.Path 对象,其实就是代表文件路径的,如果识别到是这种类型,它会自动创建对应路径的日志文件并将日志输出进去。

  • sink 可以是一个方法,可以自行定义输出实现。

  • sink 可以是一个 logging 模块的 Handler,比如 FileHandler、StreamHandler 等等,这样就可以实现自定义 Handler 的配置。

  • levelintstr,可选)– 应从中将记录的消息发送到接收器的最低严重性级别。

  • formatstr可调用,可选)– 用于在将记录的消息发送到接收器之前设置其格式的模板。

  • filter可调用strdict,可选)– 一个指令,可选择用于决定是否应将其发送到接收器的每条记录的消息。

  • colorizebool,可选)– 格式化消息中包含的颜色标记是否应转换为 ansi 代码以进行终端着色,还是以其他方式剥离。如果 ,则根据汇是不是 tty 来自动做出选择。None

  • serializebool,可选)– 在将记录的消息及其记录发送到接收器之前,是否应首先将其转换为 JSON 字符串。

  • backtracebool,可选)– 是否应向上扩展格式化的异常跟踪,超出捕获点,以显示生成错误的堆栈跟踪。

  • diagnosebool,可选)– 异常跟踪是否应显示变量值以简化调试。这应在生产中设置为,以避免泄漏敏感数据。False

  • enqueuebool,可选)– 要记录的消息在到达接收器之前是否应首先通过多进程安全队列。这在通过多个进程记录到文件时很有用。这还具有使日志记录调用非阻塞的优点。

  • catchbool,可选)– 是否应自动捕获接收器处理日志消息时发生的错误。如果 ,则会在 sys.stderr 上显示异常消息,但该异常不会传播到调用方,从而防止应用崩溃。True

  • **kwargs – 仅对配置协程或文件接收器有效的其他参数(见下文)。

如果当接收器(sink)是文件路径( pathlib.Path )时,可以应用下列参数,同时add() 会返回与所添加的接收器相关联的标识符:

    def add(
        self,
        sink: Union[str, PathLike[str]],
        *,
        level: Union[str, int] = ...,
        format: Union[str, FormatFunction] = ...,
        filter: Optional[Union[str, FilterFunction, FilterDict]] = ...,
        colorize: Optional[bool] = ...,
        serialize: bool = ...,
        backtrace: bool = ...,
        diagnose: bool = ...,
        enqueue: bool = ...,
        catch: bool = ...,
        rotation: Optional[Union[str, int, time, timedelta, RotationFunction]] = ...,
        retention: Optional[Union[str, int, timedelta, RetentionFunction]] = ...,
        compression: Optional[Union[str, CompressionFunction]] = ...,
        delay: bool = ...,
        mode: str = ...,
        buffering: int = ...,
        encoding: str = ...,
        **kwargs: Any

rotation:分隔日志文件,何时关闭当前日志文件并启动一个新文件的条件,;例如,"500 MB"、"0.5 GB"、"1 month 2 weeks"、"10h"、"monthly"、"18:00"、"sunday"、"monday at 18:00"、"06:15"
retention (str, int, datetime.timedelta or callable, optional) ,可配置旧日志的最长保留时间,例如,"1 week, 3 days"、"2 months"
compression (str or callable, optional) :日志文件在关闭时应转换为的压缩或归档格式,例如,"gz"、"bz2"、"xz"、"lzma"、"tar"、"tar.gz"、"tar.bz2"、"tar.xz"、"zip"
delay (bool, optional):是否应该在配置了接收器之后立即创建文件,或者延迟到第一个记录的消息。默认为' False '。
mode (str, optional) :与内置open()函数一样的打开模式。默认为' "a"(以附加模式打开文件)。
buffering (int, optional) :内置open()函数的缓冲策略,它默认为1(行缓冲文件)。
encoding (str, optional) :文件编码与内置的' open() '函数相同。如果' None ',它默认为'locale.getpreferredencoding() 。

基本参数

下面我们再了解下它的其他参数,例如 format、filter、level 等等。 其实它们的概念和格式和 logging 模块都是基本一样的了,例如这里使用 format、level 来规定输出的格式:

from loguru import logger


def test():
    print("Welcome to my blog--不予の測試筆記!")
    logger.add('runtime.log', format="{time:YYYY-MM-DD HH:mm:ss}  {level} From {module}.{function} : {message}", level="ERROR")
    logger.error('Welcome to my blog--不予の測試筆記!1')
    logger.warning('Welcome to my blog--不予の測試筆記!2')


if __name__ == '__main__':
    test()

控制台:
image-1655211775844
日志文件:
image-1655211795646

sink的删除操作

remove(handler_id=None)

添加 sink 之后我们也可以对其进行删除,相当于重新刷新并写入新的内容。

删除的时候根据刚刚 add 方法返回的 id 进行删除即可,看下面的例子:

# -*- coding: utf-8 -*-
from loguru import logger


i = logger.add('runtime.log', format="{message}")
logger.info("Logging")
logger.remove(i)
logger.info("No longer logging")

看这里,我们首先 add 了一个 sink,然后获取它的返回值,赋值为 i 。随后输出了一条日志,然后将i 变量传给remove 方法,再次输出一条日志,看看结果是怎样的。

控制台输出如下:

2022-06-15 09:27:44.038 | INFO     | __main__:<module>:6 - Logging
2022-06-15 09:27:44.038 | INFO     | __main__:<module>:8 - No longer logging

日志文件 runtime.log 内容如下:

Logging

可以发现,在调用 remove 方法之后,确实将历史 log 删除了。但实际上这并不是删除,只不过是将 sink 对象移除之后,在这之前的内容不会再输出到日志中。 这样我们就可以实现日志的刷新重新写入操作。

rotation配置

用了 loguru 我们还可以非常方便地使用 rotation 配置,比如我们想一天输出一个日志文件,或者文件太大了自动分隔日志文件,我们可以直接使用 add 方法的 rotation 参数进行配置。 我们看看下面的例子:

logger.add('runtime_{time}.log', rotation="500 MB")

通过这样的配置我们就可以实现每 500MB 存储一个文件,每个 log 文件过大就会新创建一个 log 文件。我们在配置 log 名字时加上了一个 time 占位符,这样在生成时可以自动将时间替换进去,生成一个文件名包含时间的 log 文件。 另外我们也可以使用 rotation 参数实现定时创建 log 文件,例如:

logger.add('runtime_{time}.log', rotation='00:00')

这样就可以实现每天 0 点新创建一个 log 文件输出了。 另外我们也可以配置 log 文件的循环时间,比如每隔一周创建一个 log 文件,写法如下:

logger.add('runtime_{time}.log', rotation='1 week')

这样我们就可以实现一周创建一个 log 文件了。

retention 配置

很多情况下,一些非常久远的 log 对我们来说并没有什么用处了,它白白占据了一些存储空间,不清除掉就会非常浪费。retention 这个参数可以配置日志的最长保留时间。 比如我们想要设置日志文件最长保留 10 天,可以这么来配置:

logger.add('runtime.log', retention='10 days')

Examples: “1 week, 3 days”, “2 months”

这样 log 文件里面就会保留最新 10 天的 log,妈妈再也不用担心 log 沉积的问题啦。

compression配置

loguru 还可以配置文件的压缩格式,比如使用 zip 文件格式保存,示例如下:

logger.add('runtime.log', compression='zip')

对应于压缩或存档的文件扩展名。这可以是以下之一: “gz”, “bz2”, “xz”, “lzma”, “tar”, “tar.gz”, “tar.bz2”,“tar.xz”, “zip”

enqueue配置

loguru可以配置在多进程同时往日志文件写日志的时候使用队列达到异步功效。

logger.add("somefile.log", enqueue=True)  # 异步写入

要记录的消息是否应在到达接收器之前首先通过多进程安全队列。这在通过多个进程记录到文件时很有用。这还具有使日志记录调用非阻塞的优点。

字符串格式化

loguru 在输出 log 的时候还提供了非常友好的字符串格式化功能,像这样:

logger.info('If you are using Python {}, prefer {feature} of course!', 3.9, feature='f-strings')

输出:

2020-05-03 10:24:34.200 | INFO     | __main__:<module>:3 - If you are using Python 3.9, prefer f-strings of course!

异常捕获

catch装饰器 方法

在很多情况下,如果遇到运行错误,而我们在打印输出 log 的时候万一不小心没有配置好 Traceback 的输出,很有可能我们就没法追踪错误所在了。

但用了 loguru 之后,我们用它提供的装饰器就可以直接进行 Traceback 的记录,类似这样的配置即可:

from loguru import logger


logger.add("runtime.log")
@logger.catch
def my_function(x, y, z):
    # An error? It's caught anyway!
    return 1 / (x + y + z)


my_function(0, 0, 0)

我们做个测试,我们在调用时三个参数都传入 0,直接引发除以 0 的错误,看看会出现什么情况
运行完毕之后,可以发现 log 里面就出现了 Traceback 信息,而且给我们输出了当时的变量值,真的是不能再赞了!结果如下:

2022-06-15 10:00:52.343 | ERROR    | __main__:<module>:11 - An error has been caught in function '<module>', process 'MainProcess' (11060), thread 'MainThread' (3460):
Traceback (most recent call last):

> File "D:/红盒子/校验\日志.py", line 11, in <module>
    my_function(0, 0, 0)
    └ <function my_function at 0x0000021185F55AF8>

  File "D:/红盒子/校验\日志.py", line 8, in my_function
    return 1 / (x + y + z)
                │   │   └ 0
                │   └ 0
                └ 0

ZeroDivisionError: division by zero

exception 方法

通过exception方法也可以实现异常的捕获与记录:

# -*- coding: utf-8 -*-
from loguru import logger

logger.add("runtime.log")


def my_function1(x, y, z):
    try:
        return 1 / (x + y + z)
    except ZeroDivisionError:
        logger.exception("What?!")


my_function1(0, 0, 0)

记录的日志信息如下所示:

2022-06-15 10:04:15.692 | ERROR    | __main__:my_function1:11 - What?!
Traceback (most recent call last):

  File "D:/红盒子/校验\日志.py", line 14, in <module>
    my_function1(0, 0, 0)
    └ <function my_function1 at 0x000001EE606F2D38>

> File "D:/红盒子/校验\日志.py", line 9, in my_function1
    return 1 / (x + y + z)
                │   │   └ 0
                │   └ 0
                └ 0

ZeroDivisionError: division by zero

Loguru简单的封装使用

# -*- coding: utf-8 -*-
"""
loguru 封装类,导入即可直接使用
# 当前文件名 logger.py
"""

from functools import wraps
import os
import datetime
import loguru


# 单例类的装饰器
def singleton_class_decorator(cls):
    """
    装饰器,单例类的装饰器
    """
    # 在装饰器里定义一个字典,用来存放类的实例。
    _instance = {}

    # 装饰器,被装饰的类
    @wraps(cls)
    def wrapper_class(*args, **kwargs):
        # 判断,类实例不在类实例的字典里,就重新创建类实例
        if cls not in _instance:
            # 将新创建的类实例,存入到实例字典中
            _instance[cls] = cls(*args, **kwargs)
        # 如果实例字典中,存在类实例,直接取出返回类实例
        return _instance[cls]

    # 返回,装饰器中,被装饰的类函数
    return wrapper_class


@singleton_class_decorator
class Logger:
    def __init__(self):
        self.logger_add()

    def get_project_path(self, project_path=None):
        if project_path is None:
            # 当前项目文件的,绝对真实路径
            # 路径,一个点代表当前目录,两个点代表当前目录的上级目录
            project_path = os.path.realpath('.')
        # 返回当前项目路径
        return project_path

    def get_log_path(self):
        # 项目目录
        project_path = self.get_project_path()
        # 项目日志目录
        project_log_dir = os.path.join(project_path, 'log')
        # 日志文件名
        project_log_filename = 'runtime_{}.log'.format(datetime.date.today())
        # 日志文件路径
        project_log_path = os.path.join(project_log_dir, project_log_filename)
        # 返回日志路径
        return project_log_path

    def logger_add(self):
        loguru.logger.add(
            sink=self.get_log_path(),
            # 日志创建周期
            rotation='00:00',
            # 保存
            retention='1 year',
            # 文件的压缩格式
            compression='zip',
            # 编码格式
            encoding="utf-8",
            # 具有使日志记录调用非阻塞的优点
            enqueue=True
        )
    # 加了@property后,可以用调用属性的形式来调用方法,后面不需要加()。
    @property
    def get_logger(self):
        return loguru.logger


'''
# 实例化日志类
'''
logger = Logger().get_logger


if __name__ == '__main__':
    logger.debug('调试代码')
    logger.info('输出信息')
    logger.success('输出成功')
    logger.warning('错误警告')
    logger.error('代码错误')
    logger.critical('崩溃输出')

    """
    在其他.py文件中,只需要直接导入已经实例化的logger类即可
    例如导入方式如下:
    from project.logger import logger
    然后直接使用logger即可
    """
    logger.info('----原始测试----')

或者使用

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2022/1/12 20:41
# @Author  : shisuiyi
# @File    : main.py
# @Software: win10 Tensorflow1.13.1 python3.9
import os

import pytest
from loguru import logger

import setting


def run(web: bool = True):

    # 解决 issues 句柄无效
    logger.remove()
    logger.add(**setting.log_filename)
    pytest.main(["-m=login", f'--alluredir={setting.REPORT_CONFIG}/data'])
    if web:
        # 自动以服务形式打开报告
        os.system(f'allure serve {setting.REPORT_CONFIG}/data')
    else:
        # 本地生成报告
        os.system(
            f'allure generate {setting.REPORT_CONFIG}/data -o {setting.REPORT_CONFIG}/html --clean'
        )
        logger.success("报告已生成")


if __name__ == "__main__":
    run()
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2022/1/20 20:31
# @Author  : shisuiyi
# @File    : setting.py
# @Software: win10 Tensorflow1.13.1 python3.9、
import os

# 项目根目录
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

# 日志目录
log_filename = {
    'sink': os.path.join(BASE_DIR, 'logs', 'run{time}.log'),
    'format': '{time} {level} {message}',
    'rotation': '5 MB',
    'enqueue': True,
    'encoding': "utf-8"
}

# 测试报告
REPORT_CONFIG = os.path.join(BASE_DIR, 'reports')
1

评论区