Loguru 实战指南:一个 add() 搞定的 Python 日志

Loguru 是 Python 最受欢迎的第三方日志库,开箱即用、零配置。本文讲解 Loguru 的 add() 配置、serialize JSON 输出、bind/contextualize 上下文、自动轮转与压缩、@catch 异常捕获,以及接入观测云集中分析的路径。

最佳实践
Loguru 实战指南:一个 add() 搞定的 Python 日志技术指南封面

Loguru 是 Python 生态中常用的第三方日志库,核心理念是"消灭配置样板代码"——一个预配置好的全局 logger,加一行 logger.add() 就能完成格式、轮转、压缩、级别的全部定制。 如果说标准库 logging 是手动挡,Loguru 就是自动挡。

核心要点速览

  • 开箱即用:from loguru import logger 直接开打,默认输出到 stderr、级别 DEBUG(可由环境配置改变),是否着色取决于终端与 colorize 设置;
  • 一切配置走 logger.add():目标、格式、级别、轮转(rotation)、保留(retention)、压缩一行搞定;
  • JSON 输出加 serialize=True;上下文用 bind()(返回带绑定字段的 logger)或 contextualize()(作用域绑定);
  • 异常捕获有神器 @logger.catch:装饰器自动记录完整堆栈;
  • 集中采集前固定 JSON 字段结构,并检查时间、级别和请求标识的提取结果。

快速上手

pip install loguru
from loguru import logger

logger.debug("调试信息")
logger.info("服务启动成功")
logger.success("任务完成")     # Loguru 独有的 SUCCESS 级别
logger.warning("配置缺失,使用默认值")
logger.error("数据库连接失败")

无需任何配置,输出即带时间戳、级别、模块、函数、行号的彩色记录。Loguru 在标准级别之外还提供更细的 TRACE(5)与 SUCCESS(25)。

add():一个方法管所有事

先移除默认 handler(logger.remove()),再按需添加:

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO", serialize=True)  # JSON 输出到 stdout
logger.add(
    "logs/app_{time}.log",
    level="DEBUG",
    rotation="100 MB",      # 按大小轮转,也支持 "1 day"、"12:00" 等
    retention="7 days",     # 在轮转/关闭等生命周期点清理,不是独立定时器
    compression="gz",       # 历史文件 gzip 压缩
    serialize=True,
)

文件轮转、保留、压缩全部内置——这是 Loguru 相比标准库最直观的生产力提升。

生产化 JSON:默认太长?自己定义

serialize=True 的默认 JSON 很详尽(含 elapsed、thread、file 等全部信息),想要精简格式可用 patch 自定义序列化:

import json
import sys
import traceback

def serialize(record):
    subset = {
        "time": record["time"].strftime("%Y-%m-%dT%H:%M:%S%z"),
        "level": record["level"].name,
        "message": record["message"],
        "extra": {k: v for k, v in record["extra"].items() if k != "serialized"},
    }
    if record["exception"]:
        exc = record["exception"]
        subset["exception"] = "".join(traceback.format_exception(exc.type, exc.value, exc.traceback))
    record["extra"]["serialized"] = json.dumps(subset, ensure_ascii=False)
    return record

logger = logger.patch(serialize)
logger.remove()
logger.add(sys.stdout, format=lambda record: "{extra[serialized]}\n", diagnose=False)

自定义 extra 值须为可 JSON 序列化类型;其他类型需明确转换策略。异常作为 JSON 字符串转义,避免追加多行破坏逐行解析。

上下文:bind 与 contextualize

import uuid

# bind:返回带固定字段的 logger,不改变其他 logger
req_logger = logger.bind(request_id="req-abc-123", user_id="USR-9")
req_logger.info("订单创建成功")   # 自动携带两个字段

# contextualize:当前上下文作用域内自动带字段,退出后还原
# 以下仅演示同步中间件,异步中间件需 await 下游调用
def logging_middleware(get_response):
    def middleware(request):
        with logger.contextualize(request_id=str(uuid.uuid4())):
            return get_response(request)
    return middleware

异常处理:@catch 装饰器

@logger.catch(reraise=True)
def risky_operation():
    return 1 / 0

默认 catch 会抑制捕获的异常;本例 reraise=True 记录后重新抛出。生产 sink 建议 diagnose=False,避免增强堆栈泄露局部变量。也可用 logger.exception() 手动记录。

一个要补的课:拦截标准库日志

第三方库(requests、uvicorn 等)大多用标准库 logging,默认不会进 Loguru 管道。需要加一个 InterceptHandler 把它们重定向过来:

import logging
import inspect

class InterceptHandler(logging.Handler):
    def emit(self, record):
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno
        frame, depth = inspect.currentframe(), 0
        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

接入观测云

文件输出可由 DataKit 的日志采集器读取:将 logfiles 指向实际日志路径,并设置 source 和 service。使用上面的自定义 JSON 时,在 Pipeline 中提取 time、level、message 和 extra.request_id,把级别映射到标准字段 status,并按输出格式解析时间。

先写入一条带已知 request_id 的日志,在日志查看器中检查时间、级别及请求标识是否一致,再用该标识筛选同一请求的记录。默认 serialize=True 的嵌套结构与自定义 JSON 不同,不能直接共用字段路径。

总结

Loguru 的生产配置公式:remove() 清场 → add(stdout, serialize=True) 给平台采集 → add(文件, rotation+retention+compression) 本地兜底 → InterceptHandler 收编第三方库。上线前还应测试轮转、进程退出时的日志排空和采集失败时的处理。

常见问题(FAQ)

Q:Loguru 适合大型项目吗?
适合。bind/contextualize 的上下文管理、异步安全(enqueue=True)、完善的轮转机制都能支撑大型项目;还需按负载验证队列、退出排空和多进程文件轮转;InterceptHandler 只接收实际传播到它的记录,不会自动保留所有自定义 extra。

Q:serialize=True 的 JSON 字段太多怎么办?
用 logger.patch() 自定义序列化函数,只保留 time/level/message/extra 等需要的字段;字段越少,采集与存储成本越低。

Q:多进程/异步环境 Loguru 安全吗?
enqueue=True 使用队列和消费线程写出,不是自动建立子进程。多进程需协调 logger/sink 的继承、multiprocessing context 与关闭顺序;不要让独立进程各自轮转同一文件。协程 sink 还需正确事件循环并 await logger.complete()。

Q:从标准库 logging 迁移到 Loguru 麻烦吗?
官方提供迁移指南;平滑做法是先用 InterceptHandler 把标准库调用桥接进 Loguru,再逐步把业务代码改写为 Loguru 调用。


系列阅读:Python 日志模块详解 | Structlog 实战指南

获取专属方案

联系我们

加入社区

微信扫码
加入官方交流群

立即体验

在线开通,按量计费,真正的云服务!

立即开始

选择观测云版本

代码托管平台