FastAPI 使用 loguru 处理日志

安装依赖

uv add loguru

配置类

from __future__ import annotations

import logging
import os
import sys
from contextvars import ContextVar
from pathlib import Path

from loguru import logger

# ----------------------------
# 1) trace_id: 用 contextvars 存“当前请求的 trace_id”
# ----------------------------
trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="-")


class InterceptHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        try:
            level = logger.level(record.levelname).name
        except Exception:
            level = record.levelno
        # depth=6 这种值可能要按你项目栈深微调
        logger.opt(depth=6, exception=record.exc_info).log(level, record.getMessage())


def get_trace_id() -> str:
    return trace_id_ctx.get()

# ----------------------------
# 2) 配置 Loguru：Console + 异步落盘 + trace_id 注入
# ----------------------------
def setup_loguru() -> None:
    log_dir = Path(os.getenv("LOG_DIR", "./logs"))
    log_dir.mkdir(parents=True, exist_ok=True)

    # 关键：移除默认 handler，自己加 console + file
    logger.remove()

    # 2.1 Console 输出（同步即可）
    logger.add(
        sys.stdout,
        level=os.getenv("LOG_LEVEL", "INFO"),
        backtrace=True,
        diagnose=False,
        colorize=True,
        format=(
            "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> "
            "<level>{level: <8}</level> "
            "[trace=<cyan>{extra[trace_id]}</cyan>] "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        ),
    )

    # 2.2 文件落盘（异步）：enqueue=True 会把写文件放到后台线程，不阻塞主流程
    # 注意：多进程/多 worker 场景下，建议加 pid 避免多个进程抢同一个文件（更稳）
    pid = os.getpid()
    logger.add(
        str(log_dir / f"app_{pid}.log"),
        level=os.getenv("LOG_LEVEL", "INFO"),
        enqueue=True,                         # 异步落盘
        rotation="00:00",                     # 每天轮转
        retention=os.getenv("LOG_RETENTION", "14 days"),  # 保留 14 天
        compression=os.getenv("LOG_COMPRESS", "gz"),      # 压缩旧日志
        backtrace=True,
        diagnose=False,
        format=(
            "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} "
            "| trace={extra[trace_id]} "
            "| {name}:{function}:{line} - {message}"
        ),
    )

    # 2.3 “自动注入 trace_id”：用 patcher 给每条日志 record 的 extra 填 trace_id
    # 这样你代码里只要 logger.info(...)，不用每次 bind/extra
    def patch_trace_id(record):
        record["extra"]["trace_id"] = get_trace_id()

    logger.configure(patcher=patch_trace_id)

    # 把标准 logging 的 root logger 指向 InterceptHandler
    logging.root.handlers = [InterceptHandler()]
    logging.root.setLevel(logging.INFO)

使用

在main.py中调用setup_loguru()即可。