日志与错误处理

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

一、logging 配置

logging.basicConfig 适合小脚本。FastAPI 项目里用字典配置更灵活——文件按天轮转、控制台和文件同时输出、不同级别分开：

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_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)
    root_logger.addHandler(file_handler)
    root_logger.addHandler(console_handler)

TimedRotatingFileHandler 按时间轮转日志文件——when="midnight" 每天午夜切一个新文件，backupCount=30 保留 30 天的旧日志。不轮转的日志文件会一直增长，把磁盘撑满。

用的时候在每个模块里拿 logger：

import logging

logger = logging.getLogger(__name__)  # __name__ 是模块名，日志里能看到来源


@app.post("/articles")
def create_article(article: ArticleCreate, db: Session = Depends(get_db)):
    logger.info("创建文章 title=%s user_id=%s", article.title, current_user_id)

    new_article = Article(title=article.title, content=article.content)
    db.add(new_article)
    db.commit()

    logger.info("文章创建成功 id=%s", new_article.id)
    return {"id": new_article.id}

logger.info("msg %s", value) 用占位符而不是 f-string——日志级别被过滤时（比如设成 WARNING，INFO 不输出），占位符的值不会被格式化，省一点开销。更重要的是，日志平台能按字段解析。

日志级别

级别	什么时候用
DEBUG	调试细节（变量值、SQL 语句）
INFO	正常流程（请求开始、操作完成）
WARNING	不正常但不影响运行（磁盘快满、重试了一次）
ERROR	出错了（接口失败、异常捕获）
CRITICAL	系统级故障（数据库连不上、服务不可用）

生产环境通常设 INFO——DEBUG 太多，ERROR 太少。

二、结构化日志

文本日志人能读，但日志平台（ELK、Grafana Loki）按字段搜索和聚合时，解析非结构化文本很费劲。结构化日志输出 JSON，每个字段直接可查：

import json
import logging


class JsonFormatter(logging.Formatter):
    """把日志格式化成 JSON。"""

    def format(self, record):
        log_data = {
            "time": self.formatTime(record),
            "level": record.levelname,
            "module": record.module,
            "message": record.getMessage(),
        }

        # 异常信息
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data, ensure_ascii=False)


# 换成 JSON 格式
file_handler.setFormatter(JsonFormatter())

输出长这样：

{"time": "2024-06-01 10:30:15", "level": "INFO", "module": "main", "message": "创建文章 title=Python 入门"}

日志平台收到这种格式，直接按 level、module、message 字段查询和过滤，不用写正则解析。

三、全局异常处理

接口里可能抛各种异常——数据库连接失败、参数缺失、业务逻辑错误。15 篇提了 BusinessError，这里做更完整的方案。

业务异常和系统异常分开处理

from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError


# 业务异常：预期的错误，返回给前端有意义的信息
class BusinessError(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


# 业务异常处理器
@app.exception_handler(BusinessError)
async def handle_business_error(request: Request, exc: BusinessError):
    logger.warning("业务异常 path=%s code=%s message=%s", request.url.path, exc.code, exc.message)
    return JSONResponse(
        status_code=200,
        content={"code": exc.code, "message": exc.message},
    )


# 请求校验异常处理器（Pydantic 校验失败时触发）
@app.exception_handler(RequestValidationError)
async def handle_validation_error(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"code": 422, "message": "参数校验失败", "detail": exc.errors()},
    )


# 兜底异常处理器：所有未捕获的异常
@app.exception_handler(Exception)
async def handle_unexpected_error(request: Request, exc: Exception):
    # 记录完整堆栈到日志，方便排查
    logger.error("未预期异常 path=%s", request.url.path, exc_info=True)

    # 返回给前端的是通用错误信息，不暴露堆栈
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误"},
    )

三层处理：业务异常（用户操作不当）→ 校验异常（参数格式不对）→ 兜底异常（代码 bug 或系统故障）。兜底处理器里 exc_info=True 把完整堆栈记到日志，但返回给前端的只有"服务器内部错误"——Python 堆栈暴露给前端既有安全隐患，也帮不了用户。

在业务逻辑里用

@app.post("/articles")
def create_article(article: ArticleCreate, db: Session = Depends(get_db)):
    # 业务校验：标题不能跟已有文章重复
    existing = db.query(Article).filter(Article.title == article.title).first()
    if existing:
        raise BusinessError(code=1001, message="标题已存在")

    # 正常逻辑
    new_article = Article(title=article.title, content=article.content)
    db.add(new_article)
    db.commit()

    return {"code": 0, "data": {"id": new_article.id}}

业务校验不通过抛 BusinessError，全局处理器统一转成 {"code": 1001, "message": "标题已存在"}。接口函数里不用写 try/except，异常处理全交给全局处理器——代码干净，格式统一。