chendandan_homework/app/logging_config.py

from __future__ import annotations

"""统一日志配置与trace_id上下文管理：通过contextvars实现一次工作流调用的trace_id贯穿所有日志。"""

import logging
import sys
import uuid
from contextvars import ContextVar


_trace_id_context: ContextVar[str] = ContextVar(
    "trace_id",
    default="-",
)


class TraceIdFilter(logging.Filter):
    """向每条日志中注入当前请求trace_id。"""

    def filter(
        self,
        record: logging.LogRecord,
    ) -> bool:
        """向日志记录注入当前上下文的trace_id，供日志格式串引用。"""

        record.trace_id = get_trace_id()
        return True


def configure_logging(
    level: int = logging.INFO,
) -> None:
    """配置TravelMind统一日志格式。"""

    handler = logging.StreamHandler(
        sys.stdout
    )

    handler.addFilter(TraceIdFilter())

    formatter = logging.Formatter(
        fmt=(
            "%(asctime)s "
            "| %(levelname)s "
            "| trace_id=%(trace_id)s "
            "| %(name)s "
            "| %(message)s"
        ),
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    handler.setFormatter(formatter)

    root_logger = logging.getLogger()

    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel(level)

    # 降低第三方HTTP库的日志噪声。
    logging.getLogger(
        "httpcore"
    ).setLevel(logging.WARNING)

    logging.getLogger(
        "httpx"
    ).setLevel(logging.WARNING)


def new_trace_id() -> str:
    """生成并设置一次工作流调用的trace_id。"""

    trace_id = uuid.uuid4().hex[:16]

    _trace_id_context.set(trace_id)

    return trace_id


def set_trace_id(
    trace_id: str,
) -> None:
    """设置外部传入的trace_id。"""

    normalized = trace_id.strip()

    _trace_id_context.set(
        normalized or "-"
    )


def get_trace_id() -> str:
    """读取当前上下文trace_id。"""

    return _trace_id_context.get()