带上下文的结构化 JSON 日志

功能说明

示例自定义 JsonFormatter，把每条日志输出成一行 JSON，包含 UTC 时间、级别、日志器名称、消息和请求 ID。请求 ID 存放在 ContextVar 中，异步任务之间互不串扰；extra={"context": ...} 里的字段会合并进 JSON，名称命中敏感词的值统一替换成 ***。异常日志额外带上格式化后的堆栈。

关键代码

payload: dict[str, object] = {
    "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
    "level": record.levelname,
    "logger": record.name,
    "message": record.getMessage(),
    "request_id": request_id.get(),
}
...
return json.dumps(payload, ensure_ascii=False, default=str)

default=str 保证日期、Decimal 等非 JSON 类型不会让日志调用本身抛错。

使用注意

• 只对 context 字典的顶层键做敏感词匹配，嵌套结构需要递归处理。
• configure_logger 会清空同名日志器已有的处理器，避免重复输出。
• 输出到标准输出，适合容器采集；写入文件时需要自行处理轮转。
• ContextVar 在 asyncio 任务中会自动复制上下文，但线程池不会自动传播。
• 日志中不要记录完整请求体或凭证，脱敏规则应集中维护。

验证方式

运行后每行都应是合法 JSON：第一条包含 order_id 和脱敏后的 token，第二条包含 error 堆栈且级别为 ERROR。可以把输出重定向到文件，再用 python -m json.tool 逐行解析验证。