FastAPI异步结构化日志实现：请求上下文与每日日志轮转

核心需求 #

• 自动捕获请求ID、路径、HTTP方法等请求上下文
• 异步日志写入，避免阻塞FastAPI事件循环
• 每日轮转日志文件，日志以JSON格式存储，便于监控系统采集
• 未处理异常日志自动携带完整请求上下文，堆栈信息格式清晰

当前实现的问题 #

• 日志写入为阻塞式，影响接口响应性能
• 请求上下文需手动拼接，无法自动附加到所有日志条目
• 日志为非结构化文本，难以被ELK等监控工具解析
• 异常堆栈信息在日志中格式混乱，不利于问题排查

解决方案 #

结合structlog（上下文管理+结构化日志）、Python标准logging模块（队列异步处理）、python-json-logger（JSON格式化）实现需求，以下是完整实现代码：

1. 依赖安装 #

pip install structlog python-json-logger

2. 完整实现代码 #

import logging
import uuid
from queue import Queue
from logging.handlers import QueueHandler, QueueListener, TimedRotatingFileHandler
from pythonjsonlogger import jsonlogger
import structlog
from contextvars import ContextVar
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 上下文变量，用于存储请求级别的日志上下文
request_context: ContextVar[dict] = ContextVar("request_context", default={})

# 1. 配置异步日志队列：将日志发送到队列，由独立线程处理写入，避免阻塞事件循环
log_queue = Queue(-1)
file_handler = TimedRotatingFileHandler(
    "logs/app.log",
    when="midnight",
    backupCount=7,
    encoding="utf-8"
)

# 2. 配置JSON格式化：包含时间戳、日志级别、上下文、消息、堆栈信息
json_formatter = jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(message)s %(exc_info)s",
    rename_fields={"asctime": "timestamp", "levelname": "level"},
    json_ensure_ascii=False
)
file_handler.setFormatter(json_formatter)

# 启动队列监听器，独立线程处理日志写入
queue_listener = QueueListener(log_queue, file_handler)
queue_listener.start()

# 3. 配置structlog：绑定上下文变量，自动注入请求上下文
structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        # 从上下文变量中加载请求上下文
        lambda _, __, event_dict: {**event_dict, **request_context.get()},
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.stdlib.render_to_log_kwargs,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# 获取structlog实例
logger = structlog.get_logger()

# 将根日志器的handler替换为QueueHandler，所有日志都会进入队列
root_logger = logging.getLogger()
root_logger.addHandler(QueueHandler(log_queue))
root_logger.setLevel(logging.INFO)

# 4. 请求中间件：绑定请求上下文到contextvar
@app.middleware("http")
async def add_request_context(request: Request, call_next):
    request_id = str(uuid.uuid4())
    # 绑定请求上下文到当前请求的上下文变量
    token = request_context.set({
        "request_id": request_id,
        "path": request.url.path,
        "method": request.method
    })
    # 将request_id返回给客户端，便于排查问题
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    # 清理上下文变量，避免请求之间的上下文污染
    request_context.reset(token)
    return response

# 5. 全局异常处理器：自动携带请求上下文记录异常
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(
        "unhandled_exception",
        error=str(exc),
        exc_info=exc
    )
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error", "request_id": request_context.get().get("request_id", "unknown")}
    )

# 示例接口：测试日志上下文自动注入
@app.get("/test")
async def test_log():
    logger.info("test_request_processed", extra_data="some_data")
    return {"message": "success"}

关键实现说明 #

• 异步日志处理：通过QueueHandler和QueueListener将日志写入操作放到独立线程，完全不阻塞FastAPI的事件循环
• 自动上下文传递：使用contextvars存储请求上下文，structlog处理器自动将上下文注入所有日志条目，无需手动拼接
• JSON结构化：python-json-logger将日志格式化为标准JSON，包含timestamp、level、request_id、path、method等字段，异常堆栈信息会被格式化为可解析的字符串
• 异常处理兼容：全局异常处理器通过contextvars获取当前请求上下文，异常日志自动携带完整请求信息，堆栈信息清晰可查

日志示例 #

{
    "timestamp": "2024-05-20T12:34:56.789Z",
    "level": "INFO",
    "request_id": "a1b2c3d4-5678-90ef-ghij-klmnopqrstuv",
    "path": "/test",
    "method": "GET",
    "event": "test_request_processed",
    "extra_data": "some_data"
}

内容的提问来源于stack exchange，提问作者shalini moorthy