如何用Loguru将FastAPI(Uvicorn)日志与回溯分文件记录
问题描述
我有一个运行在Uvicorn上的FastAPI应用,需要实现以下目标:
- 捕获Uvicorn日志并通过Loguru处理(已通过InterceptHandler实现)
- 获取请求和响应体并在debug模式下记录(已通过APIRoute实现)
- 将日志输出到两个文件:
log.log:每行一条日志消息,不含回溯信息traceback.log:仅记录错误级别及以上的日志消息,包含完整回溯信息(当前未解决,寻求帮助)
已尝试的方案
- 尝试过Loguru相关配置分离回溯日志,但无法自动生效,会从任意位置获取回溯信息
- 看过Python logging禁用堆栈跟踪的方案,但不清楚如何集成到现有代码中
- 尝试过Loguru禁用异常回溯的相关配置,未能解决问题
当前代码
main.py
from fastapi import FastAPI from rest_api.logging import init_logging, BodyLoggingRoute app = FastAPI() app.add_event_handler("startup", init_logging) # 将所有Uvicorn日志重定向到Loguru app.router.route_class = BodyLoggingRoute
rest_api/logging.py
import logging from pathlib import Path from typing import Callable import json from loguru import logger from fastapi.routing import APIRoute from fastapi import Request, Response class InterceptHandler(logging.Handler): """ Loguru文档示例中的默认处理器,用于兼容标准logging """ def emit(self, record: logging.LogRecord): # 获取对应的Loguru日志级别 try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # 定位日志消息的原始调用位置 frame, depth = logging.currentframe(), 2 while 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() ) def init_logging(min_level='INFO'): """ 替换默认logging处理器为自定义处理器,初始化Loguru配置 """ def level_filter(record): return record["level"].no >= logger.level(min_level).no # 替换Uvicorn的日志处理器 intercept_handler = InterceptHandler() logging.getLogger("uvicorn.access").handlers = [intercept_handler] logging.getLogger("uvicorn").handlers = [intercept_handler] log_path = Path(__file__).parents[2].resolve() / 'logging' logger.add(str(log_path / 'log.log'), format='{time} {level} {message}', filter=level_filter, rotation='1 month', compression='zip') class BodyLoggingRoute(APIRoute): """ 自定义路由类,用于记录请求和响应体 """ def get_route_handler(self) -> Callable: original_route_handler = super().get_route_handler() async def body_logging_route_handler(request: Request) -> Response: req_body = await request.body() response = await original_route_handler(request) res_body = response.body req_body = json.loads(req_body) if req_body else {} logger.debug(f'Request Body: {req_body}') res_body = json.loads(res_body) if res_body else {} logger.debug(f'Response JSON: {res_body}') return response return body_logging_route_handler
解决方案
修改rest_api/logging.py中的init_logging函数,添加第二个日志文件配置,实现双文件分离日志:
def init_logging(min_level='INFO'): """ 替换默认logging处理器为自定义处理器,初始化Loguru配置 """ def level_filter(record): return record["level"].no >= logger.level(min_level).no # 替换Uvicorn的日志处理器 intercept_handler = InterceptHandler() logging.getLogger("uvicorn.access").handlers = [intercept_handler] logging.getLogger("uvicorn").handlers = [intercept_handler] log_path = Path(__file__).parents[2].resolve() / 'logging' log_path.mkdir(exist_ok=True) # 确保日志目录存在 # 配置log.log:仅输出消息,不含回溯 logger.add( str(log_path / 'log.log'), format='{time} {level} {message}', filter=level_filter, rotation='1 month', compression='zip', backtrace=False, # 禁用回溯信息 diagnose=False # 禁用诊断上下文信息 ) # 配置traceback.log:仅记录ERROR及以上级别,包含完整回溯 def error_filter(record): return record["level"].no >= logger.level("ERROR").no logger.add( str(log_path / 'traceback.log'), format='{time} {level} {message}\n{exception}', # 格式中包含异常回溯 filter=error_filter, rotation='1 month', compression='zip', backtrace=True, # 启用完整回溯 diagnose=True # 启用诊断上下文(可选,帮助定位错误) )
关键修改说明
- 日志目录保障:添加
log_path.mkdir(exist_ok=True),避免因目录不存在导致日志写入失败 - log.log配置:通过
backtrace=False和diagnose=False强制禁用回溯信息,仅保留基础日志内容 - traceback.log配置:
- 用
error_filter筛选出ERROR及以上级别的日志 - 格式中加入
{exception}字段,自动输出完整的异常回溯堆栈 - 启用
backtrace=True确保捕获完整的调用栈信息
- 用
- 兼容原有逻辑:保留原有的InterceptHandler配置,确保Uvicorn日志仍能正确转发到Loguru
内容的提问来源于stack exchange,提问作者Ivan Mishalkin
相关产品推荐
相关产品推荐

