告别logging!用loguru给FastAPI日志加彩色buff的5个实战技巧

告别logging!用loguru给FastAPI日志加彩色buff的5个实战技巧 告别logging用loguru给FastAPI日志加彩色buff的5个实战技巧当你在深夜调试一个棘手的API问题时是否曾被单调的日志输出折磨得昏昏欲睡作为Python开发者我们早已厌倦了标准logging模块那冗长的配置和毫无生气的黑白输出。今天我要分享如何用loguru这个日志艺术家为你的FastAPI项目注入活力——不只是简单的彩色文字而是一套完整的日志美学解决方案。1. 为什么loguru是FastAPI开发者的理想选择在开始实战之前让我们先理解为什么loguru能在众多日志库中脱颖而出。传统logging模块需要你手动配置handler、formatter和filter而loguru采用一个logger搞定一切的哲学。它的核心优势在于零配置启动只需from loguru import logger就能立即使用内置彩色输出不同日志级别自动区分颜色无需额外插件线程/进程安全自动处理多线程环境下的日志写入强大的文件管理支持按时间、大小自动分割日志文件异常捕获自动记录完整的调用栈信息# 传统logging vs loguru的简单对比 import logging from loguru import logger # 传统方式需要这么多行才能基本配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) log logging.getLogger(__name__) # loguru只需这样 logger.info(Ready to go!)2. 基础配置5分钟打造专业级日志系统让我们从最基础的配置开始为FastAPI项目搭建日志骨架。创建一个logger.py文件作为日志模块的核心# logger.py import sys from pathlib import Path from loguru import logger # 日志保存路径 LOG_DIR Path(__file__).parent.parent / logs LOG_DIR.mkdir(exist_okTrue) # 清除默认配置 logger.remove() # 控制台输出配置 logger.add( sys.stdout, formatgreen{time:YYYY-MM-DD HH:mm:ss.SSS}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level, levelINFO ) # 文件输出配置 logger.add( LOG_DIR / fastapi_{time:YYYY-MM-DD}.log, rotation00:00, # 每天新建日志文件 retention7 days, # 保留最近7天日志 encodingutf-8, enqueueTrue, # 异步安全 backtraceTrue, # 异常回溯 diagnoseTrue, # 诊断信息 levelDEBUG )这个配置已经实现了彩色控制台输出开发时特别有用每日自动分割日志文件自动清理7天前的旧日志异步写入确保性能详细的异常堆栈记录3. 深度集成让loguru接管FastAPI全部日志默认情况下FastAPI使用的uvicorn服务器会自行输出访问日志导致系统中有两套日志体系。下面教你如何完全统一# logger.py (续) import logging from loguru import logger from uvicorn.config import LOGGING_CONFIG class InterceptHandler(logging.Handler): def emit(self, record): # 获取对应的Loguru级别 try: level logger.level(record.levelname).name except ValueError: level record.levelno # 找到调用源 frame logging.currentframe() depth 2 while frame.f_code.co_filename logging.__file__: frame frame.f_back depth 1 logger.opt(depthdepth, exceptionrecord.exc_info).log( level, record.getMessage() ) def setup_logging(): # 拦截uvicorn和fastapi的日志 logging.getLogger(uvicorn).handlers [InterceptHandler()] logging.getLogger(uvicorn.access).handlers [InterceptHandler()] logging.getLogger(fastapi).handlers [InterceptHandler()] # 禁用uvicorn默认配置 LOGGING_CONFIG[formatters][default][fmt] %(message)s LOGGING_CONFIG[formatters][access][fmt] %(message)s在FastAPI启动文件中应用这个配置# main.py from fastapi import FastAPI from .logger import logger, setup_logging app FastAPI() app.on_event(startup) async def startup(): setup_logging() logger.info(Application started) app.get(/) async def root(): logger.debug(This is a debug message) return {message: Hello World}现在所有日志——包括uvicorn的访问日志——都将通过loguru输出保持统一的格式和目的地。4. 高级技巧让你的日志会说话4.1 动态日志级别控制有时生产环境需要临时调整日志级别来排查问题而无需重启服务# 添加一个HTTP端点来动态修改日志级别 from fastapi import APIRouter from loguru import logger router APIRouter() router.post(/log/level/{level}) async def change_log_level(level: str): valid_levels [TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL] if level.upper() not in valid_levels: return {status: error, message: fInvalid level, choose from {valid_levels}} logger.remove() logger.add(sys.stdout, levellevel.upper()) return {status: success, new_level: level}4.2 结构化日志记录对于需要后期分析的重要信息使用JSON格式记录# 在logger.py中添加新的sink logger.add( LOG_DIR / structured_{time:YYYY-MM-DD}.json, format{message}, serializeTrue, # 自动转为JSON rotation00:00, retention30 days, levelINFO ) # 使用时 logger.bind(user_id123, endpoint/api/v1/users).info(User logged in)4.3 性能关键路径的延迟计算避免在日志语句中进行昂贵的计算使用lambda延迟执行# 不推荐 - 即使日志级别高于DEBUG也会执行expensive_operation() logger.debug(fData: {expensive_operation()}) # 推荐 - 仅在需要记录DEBUG日志时才会调用 logger.debug(Data: {}, lambda: expensive_operation())5. 生产环境最佳实践5.1 日志文件命名策略根据环境使用不同的日志策略环境文件命名规则保留策略压缩开发fastapi_dev_{time}.log7天否测试fastapi_test_{time}.log30天是生产fastapi_prod_{time}.log90天是# 根据环境配置不同 if ENV production: logger.add( LOG_DIR / fastapi_prod_{time:YYYY-MM-DD}.log, rotation100 MB, retention90 days, compressionzip )5.2 异常捕获装饰器创建一个自动记录异常的装饰器from functools import wraps from typing import Callable, Any def log_exceptions(func: Callable) - Callable: wraps(func) async def wrapper(*args, **kwargs) - Any: try: return await func(*args, **kwargs) except Exception as e: logger.opt(exceptione).error( fException in {func.__name__}: {str(e)} ) raise return wrapper # 使用示例 app.get(/items/{item_id}) log_exceptions async def read_item(item_id: int): # 你的业务逻辑5.3 请求ID追踪为每个请求添加唯一ID方便追踪from uuid import uuid4 from fastapi import Request app.middleware(http) async def add_request_id(request: Request, call_next): request_id str(uuid4()) with logger.contextualize(request_idrequest_id): response await call_next(request) response.headers[X-Request-ID] request_id return response # 现在所有日志都会自动包含request_id字段6. 调试技巧让日志成为你的超能力6.1 彩色日志的视觉提示利用loguru的颜色标签创建更直观的输出logger.add( sys.stdout, formatblue{time:HH:mm:ss}/blue | level{level.icon}/level # loguru 0.6.0支持图标 level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan - level{message}/level, levelDEBUG )6.2 关键路径标记使用logger.patch()为特定模块的日志添加标记def patcher(record): if important_module in record[file].name: record[extra][mark] return record logger logger.patch(patcher)6.3 性能分析日志结合上下文管理器记录代码块执行时间from contextlib import contextmanager import time contextmanager def log_duration(description: str): start time.monotonic() yield duration time.monotonic() - start logger.debug(f{description} took {duration:.3f}s) # 使用示例 with log_duration(Database query): # 执行数据库操作