数据STUDIO

强强联手!Loguru 和 Pydantic 设置 Python 日志

Image

Image

在分布式系统复杂度指数增长的今天(据CNCF 2023报告,生产环境平均部署8.3个可观测性工具),日志系统已从简单的调试工具演进为系统的神经中枢。Python开发者正面临新的范式转变——通过Loguru和Pydantic的现代组合,我们可以实现:

  1. 智能诊断 - 结构化日志自动捕获上下文堆栈
  2. 配置即代码 - 类型安全的Pydantic模型管理日志级别/格式
  3. 架构韧性 - 跨模块的统一日志接口设计

本文将揭示如何超越print()语句,构建符合Twelve-Factor App原则的日志体系。你将掌握:

  • Loguru[1] 的上下文感知日志(含异步处理最佳实践)
  • 基于Pydantic的环境敏感配置方案
  • 微服务场景下的分布式追踪集成

本文面向所有重视系统可观测性的技术从业者,特别是使用Python构建现代应用的开发者——无论是开发FastAPI/Flask的API工程师、设计ETL流水线的数据专家(如Airflow用户),还是追求标准化日志的DevOps/SRE团队。如果你正在寻找一种更优雅、更可维护的方式来替代散落的print()语句,这些内容将为你提供直接可用的解决方案。

为什么不直接使用logging?

没错,这个logging模块很灵活。但它也比较冗长、重复,而且除非你非常小心,否则各个模块之间会不一致。此外,轮换文件、格式化 JSON 或捕获第三方日志通常需要大量的配置。

了解Loguru:简单的语法、智能的默认值和一个一致的logger对象。结合Pydantic的设置管理和.env支持,我们现在拥有一个简洁、声明式的日志配置。

架构概述

我们将构建一个完全可重复使用的Logger Factory:

  • 面向开发人员的控制台日志记录
  • 生产管道的 JSON 文件日志记录
  • 基于 .env使用pydantic配置
  • 拦截标准和第三方日志
  • setup_logging()和get_logger()函数

该项目架构:

project/ 
├── logger_factory.py        # 核心日志设置和 get_logger()
├── intercept_handler.py     # 将 stdlib 日志重定向到 Loguru
├── settings.py              # 来自 .env 的 Pydantic 设置
├── main.py                  # 应用程序入口点
├── some_module.py           # 在模块中使用记录器的示例
└── .env                     # 集中式环境配置

步骤 1:使用Pydantic 设置.env

#settings.py 
from pydantic import BaseSettings, Field
from typing import List
classSettings(BaseSettings):
    log_level: str = Field("DEBUG", description="Minimum log level")
    log_to_console: bool = Field(True, description="Enable stderr logging")
    log_to_file: bool = Field(True, description="Enable file logging")
    log_file: str = Field("app.log", description="Path for log output")
    intercept_modules: List[str] = Field(default_factory=lambda: ["uvicorn", "sqlalchemy"])
classConfig:
        env_file = ".env"

你的.env可能看起来像:

LOG_LEVEL =INFO 
LOG_TO_CONSOLE = true 
LOG_TO_FILE = true 
LOG_FILE =logs/service.log 
INTERCEPT_MODULES =uvicorn,sqlalchemy

步骤2:将标准日志重定向到 Loguru

# intercept_handler.py
import logging
from loguru import logger
classInterceptHandler(logging.Handler):
defemit(self, record):
try:
            level = logger.level(record.levelname).name
except ValueError:
            level = record.levelno
        frame, depth = logging.currentframe(), 2
while frame and frame.f_globals.get("__name__") == __name__:
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

这使得logging.getLogger("uvicorn")任何标准logging.warning()输出都可以干净地传输到你的 Loguru 记录器中。

步骤 3:记录器工厂

from loguru import logger
from intercept_handler import InterceptHandler
from settings import Settings
import sys, logging
settings = Settings()
defsetup_logging():
    logger.remove()
    format_console = (
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan> | "
"<level>{level}</level> | <level>{message}</level>"
    )
    format_file = (
"{{\"time\":\"{time:YYYY-MM-DDTHH:mm:ss}\","
"\"level\":\"{level}\",\"module\":\"{module}\",\"
        "
message\":\"{message}\"}}"
    )
if settings.log_to_console:
        logger.add(sys.stderr, level=settings.log_level.upper(), format=format_console, enqueue=True)
if settings.log_to_file:
        logger.add(settings.log_file, level=settings.log_level.upper(), format=format_file,
                   enqueue=True, rotation="1 week", serialize=True)
    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
for mod in settings.intercept_modules:
        logging.getLogger(mod).handlers = [InterceptHandler()]
        logging.getLogger(mod).propagate = False
return logger
defget_logger():
return logger

步骤 4:在模块中使用

# some_module.py
import logging
std_logger = logging.getLogger(__name__)
from logger_factory import get_logger
loguru_logger = get_logger()
defdo_something():
    std_logger.warning("Standard logging warning")
    loguru_logger.info("Loguru says hello from module")

你不必担心模块是否使用标准logging或loguru——由于拦截处理程序,两者都将显示格式化和集中化。

设置一次

# main.py
from logger_factory import setup_logging
setup_logging()
from some_module import do_something
do_something()

或者在你的包中实现自动化__init__.py以确保日志记录始终可用。

测试

只需运行:

python main.py

你的控制台将显示人性化的日志。你的app.log文件将包含可供日志收集器提取的结构化 JSON。

写在最后

掌握日志意味着掌握可观察性。借助 Loguru、Pydantic 和简洁的架构,你可以:

  • 一次配置,随处使用
  • 通过结构化日志追踪错误
  • 在整个堆栈中清晰地路由日志
  • 随着应用程序的增长扩展日志记录策略

你刚刚用 Python 构建了一个生产级日志记录器。与你的团队分享它,重构你的应用,然后享受清晰易懂的日志吧。

参考资料
[1] 

Loguru: https://github.com/Delgan/loguru

🏴‍☠️宝藏级🏴‍☠️ 原创公众号『数据STUDIO』内容超级硬核。公众号以Python为核心语言,垂直于数据科学领域,包括可戳👉Python|MySQL|数据分析|数据可视化|机器学习与数据挖掘|爬虫等,从入门到进阶!

长按👇关注- 数据STUDIO -设为星标,干货速递ImageImage