钩子、错误处理与日志

第 9 章 · 钩子、错误处理与日志

本章目标：掌握 Flask 应用钩子（before_request / after_request / teardown_request）；使用 @app.errorhandler 统一 API 与 Web 错误页；配置 logging 分级输出与日志轮转；实现 请求 ID 贯穿全链路；编写结构化日志便于检索与排障；在 api-demo 建立可观测性基础。

学时建议：4 小时（含 1.5 小时日志跟练）

前置：本模块 ch01 请求生命周期；ch06 Flask-Login；ch07 API 错误码；ch08 分环境配置。HTTP 基础见 frontend-web ch08。

9.1 Flask 请求生命周期

HTTP Request
  → before_request（可短路返回 Response）
  → 路由匹配 → 视图函数
  → 视图返回 Response
  → after_request（可修改 Response 头）
  → teardown_request（无论是否异常都执行）

与 Django 中间件对比：

示例项目 api-demo 为教学虚构 API 服务；日志字段使用 user-demo 账号，不记录任何真实生产流量。

9.2 请求 ID 与 before_request

为每个请求生成唯一 request_id，写入 g 对象并在响应头返回，便于前后端与日志关联。

api_demo/middleware/request_id.py：

import uuid
import time
from flask import g, request


def register_request_hooks(app):
    @app.before_request
    def assign_request_id():
        g.request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:12]
        g.request_start = time.perf_counter()

    @app.after_request
    def attach_request_id(response):
        response.headers["X-Request-ID"] = g.get("request_id", "-")
        return response

在 create_app 中调用：

from api_demo.middleware.request_id import register_request_hooks

def create_app(...):
    # ...
    register_request_hooks(app)
    return app

前端传递 request_id（可选，用于分布式追踪）：

fetch("/api/v1/products", {
  headers: { "X-Request-ID": crypto.randomUUID() },
});

9.3 请求日志与耗时

api_demo/middleware/access_log.py：

import logging
import time
from flask import g, request
from flask_login import current_user

logger = logging.getLogger("api_demo.access")


def register_access_log(app):
    @app.after_request
    def log_access(response):
        start = getattr(g, "request_start", None)
        duration_ms = (time.perf_counter() - start) * 1000 if start else -1
        user = "-"
        if current_user.is_authenticated:
            user = current_user.username

        logger.info(
            "request_id=%s method=%s path=%s status=%s duration_ms=%.1f user=%s ip=%s",
            g.get("request_id", "-"),
            request.method,
            request.path,
            response.status_code,
            duration_ms,
            user,
            _client_ip(),
        )

        if duration_ms > 500:
            logging.getLogger("api_demo.slow").warning(
                "slow_request request_id=%s path=%s duration_ms=%.1f",
                g.get("request_id"),
                request.path,
                duration_ms,
            )
        return response


def _client_ip():
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.remote_addr or "-"

9.4 logging 配置

api_demo/logging_config.py：

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path


def setup_logging(app):
    log_dir = Path(app.instance_path) / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)

    level = logging.DEBUG if app.debug else logging.INFO
    formatter = logging.Formatter(
        "%(asctime)s %(levelname)s [%(name)s] %(message)s"
    )

    # 控制台
    console = logging.StreamHandler()
    console.setFormatter(formatter)
    console.setLevel(level)

    # 滚动文件
    file_handler = RotatingFileHandler(
        log_dir / "api_demo.log",
        maxBytes=5 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.INFO)

    for name in ("api_demo.access", "api_demo.slow", "api_demo.audit"):
        log = logging.getLogger(name)
        log.setLevel(logging.INFO)
        log.addHandler(console)
        log.addHandler(file_handler)
        log.propagate = False

    # 压低 werkzeug 默认噪音
    logging.getLogger("werkzeug").setLevel(logging.WARNING)

create_app 末尾：

from api_demo.logging_config import setup_logging

def create_app(...):
    # ...
    setup_logging(app)
    return app

在业务代码中使用：

import logging
audit_log = logging.getLogger("api_demo.audit")

def update_product_price(product, old_price, new_price):
    product.price = new_price
    audit_log.info(
        "price_change product_id=%s old=%s new=%s request_id=%s",
        product.id, old_price, new_price, g.get("request_id"),
    )