Py学习  »  Python

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

数据STUDIO • 1 周前 • 23 次点击  


在分布式系统复杂度指数增长的今天(据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
class Settings(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"])
    class Config:
        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
class InterceptHandler(logging.Handler):
    def emit(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()
def setup_logging():
    logger.remove()
    format_console = (
        "{time:YYYY-MM-DD HH:mm:ss} | "
        "{name}:{function} | "
        "{level} | {message}"
    )
    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
def get_logger():
    return logger

步骤 4:在模块中使用

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

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

设置一次

# 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为核心语言,垂直于数据科学领域,包括可戳👉 PythonMySQL 数据分析数据可视化机器学习与数据挖掘爬虫 等,从入门到进阶!

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

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/183885
 
23 次点击