59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
"""结构化 JSON 日志配置模块。"""
|
||
import logging
|
||
import json
|
||
from datetime import datetime, timezone
|
||
|
||
from app.middleware.logging_filter import APIKeyFilter
|
||
|
||
|
||
class JSONFormatter(logging.Formatter):
|
||
"""将日志记录格式化为 JSON 字符串，便于日志收集平台（如 ELK、Loki）解析。"""
|
||
|
||
def format(self, record: logging.LogRecord) -> str:
|
||
log_entry: dict = {
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"level": record.levelname,
|
||
"message": record.getMessage(),
|
||
"logger": record.name,
|
||
"module": record.module,
|
||
"function": record.funcName,
|
||
"line": record.lineno,
|
||
}
|
||
|
||
if record.exc_info:
|
||
log_entry["exception"] = self.formatException(record.exc_info)
|
||
|
||
# 从 extra 字段注入的可观测性上下文
|
||
if hasattr(record, "user_id"):
|
||
log_entry["user_id"] = record.user_id
|
||
if hasattr(record, "request_id"):
|
||
log_entry["request_id"] = record.request_id
|
||
if hasattr(record, "path"):
|
||
log_entry["path"] = record.path
|
||
if hasattr(record, "method"):
|
||
log_entry["method"] = record.method
|
||
if hasattr(record, "duration_ms"):
|
||
log_entry["duration_ms"] = record.duration_ms
|
||
if hasattr(record, "status_code"):
|
||
log_entry["status_code"] = record.status_code
|
||
|
||
return json.dumps(log_entry, ensure_ascii=False)
|
||
|
||
|
||
def setup_logging(level: int = logging.INFO) -> None:
|
||
"""初始化全局 JSON 日志配置。
|
||
|
||
应在应用启动时（import 其他模块之前）调用一次。
|
||
"""
|
||
handler = logging.StreamHandler()
|
||
handler.setFormatter(JSONFormatter())
|
||
|
||
root_logger = logging.getLogger()
|
||
root_logger.handlers.clear()
|
||
root_logger.addHandler(handler)
|
||
root_logger.setLevel(level)
|
||
root_logger.addFilter(APIKeyFilter())
|
||
|
||
logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
|
||
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
|