在分布式系统复杂度指数增长的今天(据CNCF 2023报告,生产环境平均部署8.3个可观测性工具),日志系统已从简单的调试工具演进为系统的神经中枢。Python开发者正面临新的范式转变——通过Loguru和Pydantic的现代组合,我们可以实现:
- 配置即代码 - 类型安全的Pydantic模型管理日志级别/格式
本文将揭示如何超越print()
语句,构建符合Twelve-Factor App原则的日志体系。你将掌握:
- Loguru[1] 的上下文感知日志(含异步处理最佳实践)
本文面向所有重视系统可观测性的技术从业者,特别是使用Python构建现代应用的开发者——无论是开发FastAPI/Flask的API工程师、设计ETL流水线的数据专家(如Airflow用户),还是追求标准化日志的DevOps/SRE团队。如果你正在寻找一种更优雅、更可维护的方式来替代散落的print()
语句,这些内容将为你提供直接可用的解决方案。
为什么不直接使用logging
?
没错,这个logging
模块很灵活。但它也比较冗长、重复,而且除非你非常小心,否则各个模块之间会不一致。此外,轮换文件、格式化 JSON 或捕获第三方日志通常需要大量的配置。
了解Loguru:简单的语法、智能的默认值和一个一致的logger
对象。结合Pydantic的设置管理和.env
支持,我们现在拥有一个简洁、声明式的日志配置。
架构概述
我们将构建一个完全可重复使用的Logger Factory:
-
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")
你不必担心模块是否使用标准
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