46｜日志配置与结构化输出

print 在开发时够用，但上线后日志需要写文件、带时间戳、按级别过滤、能被日志平台（Loki、ELK）收集。Python 标准库 logging 提供这些能力，配合 FastAPI 的全局配置，让线上问题能追溯到、错误响应统一、堆栈不泄露给前端。

一、logging 基础

日志级别

级别	数值	用途
DEBUG	10	调试信息，开发时看详细流程
INFO	20	正常运行信息，如请求处理完成
WARNING	30	警告，不立即影响功能但需要注意
ERROR	40	错误，功能受损但服务还能跑
CRITICAL	50	严重错误，服务可能崩溃

级别越高，输出的日志越少。设置级别为 INFO 时，DEBUG 日志不会输出。

基础用法

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

logger.debug("调试信息")
logger.info("用户 %s 登录成功", "admin")
logger.warning("磁盘使用率超过 80%%")
logger.error("数据库连接失败")

二、FastAPI 项目中的日志配置

basicConfig 适合小脚本，项目中用字典配置更灵活：

import logging
from logging.handlers import TimedRotatingFileHandler

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

    # 文件处理器：按天轮转，保留 30 天
    file_handler = TimedRotatingFileHandler(
        filename="logs/app.log",
        when="midnight",
        backupCount=30,
        encoding="utf-8",
    )
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.INFO)

    # 控制台处理器：开发时看输出
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    console_handler.setLevel(logging.DEBUG)

    # 配置根 logger
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(file_handler)
    root.addHandler(console_handler)

处理器	输出位置	级别	用途
TimedRotatingFileHandler	文件	INFO	线上日志收集
StreamHandler	控制台	DEBUG	开发调试

每个模块中获取 logger：

import logging

logger = logging.getLogger(__name__)   # 模块名作为 logger 名

__name__ 让日志来源清晰：app.routers.servers 表示 app/routers/servers.py 中输出的日志。

三、结构化日志

传统文本日志需要正则解析，结构化日志（JSON）让日志平台直接索引字段：

import json
import logging

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data, ensure_ascii=False)

json_handler = logging.StreamHandler()
json_handler.setFormatter(JSONFormatter())

输出：

{"timestamp": "2026-06-27 14:32:01", "level": "INFO", "logger": "app.routers.servers", "message": "查询服务器列表"}

Loki、ELK 等日志平台可以直接按 level、logger 等字段过滤和聚合。

四、请求上下文注入

把请求 ID、用户 ID 等上下文信息注入每条日志：

import contextvars
import logging

request_id_var = contextvars.ContextVar("request_id", default="-")

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

formatter = logging.Formatter(
    "%(asctime)s %(levelname)s [%(request_id)s] %(message)s"
)

logger = logging.getLogger()
logger.addFilter(ContextFilter())

在中间件中设置上下文：

@app.middleware("http")
async def add_request_context(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_var.set(request_id)
    response = await call_next(request)
    return response

五、常见错误

日志文件不轮转导致磁盘满

# 错误：FileHandler 不轮转，日志文件无限增长
logging.FileHandler("app.log")

# 正确：按时间或大小轮转
TimedRotatingFileHandler("app.log", when="midnight", backupCount=30)
RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5)

重复添加 Handler

# 错误：每次导入都添加 Handler，日志重复输出
logger = logging.getLogger(__name__)
logger.addHandler(console_handler)

# 正确：只在应用启动时配置一次

生产环境输出 DEBUG 日志

# 错误：DEBUG 日志量巨大，影响性能和磁盘
root.setLevel(logging.DEBUG)

# 正确：生产环境用 INFO 或 WARNING
root.setLevel(logging.INFO)