13-错误处理与日志

Python 3.11+

本章讲解 Flask 应用中的错误处理机制和日志管理，从基础错误响应到生产级日志方案。

第一部分：错误处理（L1）

1.1 HTTP 错误：abort() 函数

实际场景：

用户请求一个不存在的文章 ID /articles/999，你需要立即返回 404 状态码，而不是继续执行后续逻辑。

问题：如何在视图函数中主动抛出 HTTP 错误？

abort() 基本用法：

abort() 函数会立即中断请求处理流程，返回指定的 HTTP 错误状态码。

python

# article_service.py
from flask import Flask, abort
from typing import Any

app: Flask = Flask(__name__)

ARTICLES: dict[int, dict[str, Any]] = {
    1: {"id": 1, "title": "Python 入门", "content": "..."},
    2: {"id": 2, "title": "Flask 进阶", "content": "..."},
}


@app.route("/articles/<int:article_id>")
def get_article(article_id: int) -> dict[str, Any]:
    """获取文章详情"""
    article: dict[str, Any] | None = ARTICLES.get(article_id)
    if article is None:
        abort(404)  # 立即返回 404 响应
    return article

代码	含义
`abort(404)`	抛出 404 Not Found 错误
`abort(400)`	抛出 400 Bad Request 错误
`abort(500)`	抛出 500 Internal Server Error 错误

常用 HTTP 错误码：

HTTP 错误码分类：
┌──────────────────────────────────────────────────────────┐
│  4xx 客户端错误                                          │
│  ├── 400 Bad Request        请求参数错误                │
│  ├── 401 Unauthorized       未认证                      │
│  ├── 403 Forbidden          无权限                      │
│  ├── 404 Not Found          资源不存在                  │
│  └── 405 Method Not Allowed HTTP 方法不允许             │
│                                                          │
│  5xx 服务器错误                                          │
│  ├── 500 Internal Server Error  服务器内部错误           │
│  ├── 502 Bad Gateway          网关错误                   │
│  └── 503 Service Unavailable  服务不可用                 │
└──────────────────────────────────────────────────────────┘

python

# error_examples.py
from flask import Flask, abort, request
from typing import Any

app: Flask = Flask(__name__)


@app.route("/users", methods=["POST"])
def create_user() -> tuple[dict[str, str], int]:
    """创建用户 - 多种错误场景"""
    data: dict[str, Any] | None = request.get_json()

    if data is None:
        abort(400, description="请求体必须是 JSON 格式")

    if "name" not in data:
        abort(400, description="缺少必填字段：name")

    if not isinstance(data["name"], str) or len(data["name"].strip()) == 0:
        abort(400, description="name 必须是非空字符串")

    # 模拟已存在检查
    if data["name"] == "admin":
        abort(409, description="用户名已存在")

    return {"message": f"用户 {data['name']} 创建成功"}, 201

1.2 自定义错误处理器

实际场景：

Flask 默认的 404 错误页面是纯文本的，你需要返回一个美观的 HTML 错误页面。

问题：如何自定义错误页面的内容？

@app.errorhandler() 装饰器：

python

# error_handlers.py
from flask import Flask, render_template
from werkzeug.exceptions import HTTPException

app: Flask = Flask(__name__)


@app.errorhandler(404)
def not_found_error(error: Exception) -> tuple[str, int]:
    """自定义 404 页面"""
    return render_template("errors/404.html"), 404


@app.errorhandler(500)
def internal_error(error: Exception) -> tuple[str, int]:
    """自定义 500 页面"""
    return render_template("errors/500.html"), 500

错误处理器接收错误对象作为参数：

python

# advanced_error_handlers.py
from flask import Flask, jsonify, request
from werkzeug.exceptions import HTTPException, NotFound, BadRequest

app: Flask = Flask(__name__)


@app.errorhandler(NotFound)
def handle_not_found(error: NotFound) -> tuple[dict[str, str], int]:
    """处理 404 错误"""
    return {
        "error": "Not Found",
        "message": f"请求的资源不存在: {request.path}",
    }, 404


@app.errorhandler(BadRequest)
def handle_bad_request(error: BadRequest) -> tuple[dict[str, str], int]:
    """处理 400 错误"""
    return {
        "error": "Bad Request",
        "message": str(error.description),
    }, 400

1.3 错误处理器的返回值

实际场景：

错误处理器需要返回正确的响应格式，包括响应体和状态码。

问题：错误处理器可以返回哪些格式？

返回值格式说明：

错误处理器返回值格式：
┌──────────────────────────────────────────────────────────┐
│  格式一：(response, status_code)                          │
│  return render_template("404.html"), 404                 │
│                                                          │
│  格式二：response（状态码从 error 对象推断）              │
│  return jsonify({"error": "Not Found"})                  │
│                                                          │
│  格式三：(response, status_code, headers)                 │
│  return jsonify({"error": "..."}), 400, {"X-Retry": "1"} │
│                                                          │
│  格式四：Response 对象                                    │
│  return make_response(..., 404)                          │
└──────────────────────────────────────────────────────────┘

python

# response_formats.py
from flask import Flask, jsonify, make_response, render_template, Response
from typing import Any

app: Flask = Flask(__name__)


# 格式一：元组 (html, 状态码)
@app.errorhandler(403)
def forbidden_html(error: Exception) -> tuple[str, int]:
    return render_template("errors/403.html"), 403


# 格式二：纯 JSON（Flask 自动推断 200，但错误处理器会保持原状态码）
@app.errorhandler(400)
def bad_request_json(error: Exception) -> dict[str, Any]:
    return {"error": "Bad Request", "code": 400}


# 格式三：元组 (json, 状态码, 响应头)
@app.errorhandler(429)
def rate_limit_error(error: Exception) -> tuple[dict[str, Any], int, dict[str, str]]:
    return (
        {"error": "Rate Limit Exceeded", "retry_after": 60},
        429,
        {"Retry-After": "60", "X-RateLimit-Reset": "1234567890"},
    )


# 格式四：Response 对象
@app.errorhandler(503)
def service_unavailable(error: Exception) -> Response:
    response = make_response(
        render_template("errors/503.html"),
        503,
    )
    response.headers["Retry-After"] = "300"
    return response

1.4 Blueprint 级别的错误处理器

实际场景：

你的应用有多个蓝图（用户模块、管理后台、API 接口），每个模块需要不同的错误处理方式。

问题：如何为单个蓝图注册错误处理器？

python

# blueprints/api.py
from flask import Blueprint, jsonify
from typing import Any

api_bp: Blueprint = Blueprint("api", __name__, url_prefix="/api")


@api_bp.errorhandler(404)
def api_not_found(error: Exception) -> tuple[dict[str, Any], int]:
    """API 蓝图专用的 404 处理"""
    return {"error": "API_NOT_FOUND", "message": "接口不存在"}, 404


@api_bp.errorhandler(500)
def api_internal_error(error: Exception) -> tuple[dict[str, Any], int]:
    """API 蓝图专用的 500 处理"""
    return {"error": "API_INTERNAL_ERROR", "message": "服务器内部错误"}, 500

python

# blueprints/admin.py
from flask import Blueprint, render_template

admin_bp: Blueprint = Blueprint("admin", __name__, url_prefix="/admin")


@admin_bp.errorhandler(403)
def admin_forbidden(error: Exception) -> tuple[str, int]:
    """管理后台专用的 403 处理"""
    return render_template("admin/errors/403.html"), 403


@admin_bp.errorhandler(404)
def admin_not_found(error: Exception) -> tuple[str, int]:
    """管理后台专用的 404 处理"""
    return render_template("admin/errors/404.html"), 404

错误处理器优先级：
┌──────────────────────────────────────────────────────────┐
│  请求进入                                                  │
│       │                                                    │
│       ↓                                                    │
│  1. Blueprint 级别的 errorhandler（最高优先级）            │
│       │                                                    │
│       ↓                                                    │
│  2. Application 级别的 errorhandler                        │
│       │                                                    │
│       ↓                                                    │
│  3. Flask 默认错误处理（兜底）                              │
│                                                            │
│  注意：Blueprint 的错误处理器只对注册到该蓝图的路由生效      │
└──────────────────────────────────────────────────────────┘

python

# main.py
from flask import Flask
from blueprints.api import api_bp
from blueprints.admin import admin_bp

app: Flask = Flask(__name__)

# 应用级错误处理器（全局兜底）
@app.errorhandler(500)
def global_500(error: Exception) -> str:
    return "全局 500 页面"


app.register_blueprint(api_bp)
app.register_blueprint(admin_bp)

# /api/xxx 出错 → api_bp.errorhandler(404)
# /admin/xxx 出错 → admin_bp.errorhandler(403)
# 其他路由出错 → app.errorhandler(500)

第二部分：JSON API 错误（L1）

2.1 RESTful API 标准错误响应格式

实际场景：

前端调用你的 API 时，需要统一的错误格式来解析和展示错误信息。

问题：如何设计统一的 API 错误响应格式？

标准错误响应格式：

API 错误响应结构：
┌──────────────────────────────────────────────────────────┐
│  {                                                       │
│    "error": {                                            │
│      "code": "USER_NOT_FOUND",    // 业务错误码           │
│      "message": "用户不存在",       // 用户可读信息        │
│      "details": {...},           // 额外信息（可选）      │
│      "request_id": "req_xxx"     // 请求追踪 ID          │
│    }                                                     │
│  }                                                       │
└──────────────────────────────────────────────────────────┘

python

# api_errors.py
from flask import Flask, jsonify, request
from typing import Any
import uuid

app: Flask = Flask(__name__)


def make_api_error(
    code: str,
    message: str,
    status_code: int,
    details: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], int]:
    """构建统一的 API 错误响应"""
    error_response: dict[str, Any] = {
        "error": {
            "code": code,
            "message": message,
            "request_id": request.headers.get("X-Request-ID", str(uuid.uuid4())[:8]),
        }
    }
    if details is not None:
        error_response["error"]["details"] = details
    return error_response, status_code


@app.route("/users/<int:user_id>")
def get_user(user_id: int) -> tuple[dict[str, Any], int]:
    """获取用户"""
    users_db: dict[int, dict[str, str]] = {1: {"id": "1", "name": "Alice"}}
    user: dict[str, str] | None = users_db.get(user_id)

    if user is None:
        return make_api_error(
            code="USER_NOT_FOUND",
            message=f"用户 {user_id} 不存在",
            status_code=404,
        )
    return user, 200


@app.route("/users", methods=["POST"])
def create_user() -> tuple[dict[str, Any], int]:
    """创建用户 - 参数验证错误"""
    data: dict[str, Any] | None = request.get_json()

    if data is None:
        return make_api_error(
            code="INVALID_JSON",
            message="请求体必须是 JSON 格式",
            status_code=400,
        )

    errors: list[str] = []
    if "name" not in data:
        errors.append("name 是必填字段")
    if "email" not in data:
        errors.append("email 是必填字段")
    elif "@" not in data["email"]:
        errors.append("email 格式不正确")

    if errors:
        return make_api_error(
            code="VALIDATION_ERROR",
            message="参数验证失败",
            status_code=400,
            details={"field_errors": errors},
        )

    return {"message": "创建成功"}, 201

2.2 统一错误处理装饰器

实际场景：

业务逻辑中到处是 try/except，错误处理代码重复且难以维护。

问题：如何用装饰器统一处理视图函数中的异常？

python

# error_decorator.py
from flask import Flask, jsonify, request
from functools import wraps
from typing import Any, Callable
import logging
import uuid

app: Flask = Flask(__name__)
logger: logging.Logger = logging.getLogger(__name__)


def handle_api_errors(
    func: Callable[..., Any],
) -> Callable[..., tuple[dict[str, Any], int]]:
    """统一 API 错误处理装饰器"""

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> tuple[dict[str, Any], int]:
        try:
            return func(*args, **kwargs)
        except ValueError as e:
            # 参数错误
            request_id: str = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
            logger.warning(
                "参数错误 [%s]: %s - %s %s",
                request_id,
                str(e),
                request.method,
                request.path,
            )
            return {
                "error": {
                    "code": "VALIDATION_ERROR",
                    "message": str(e),
                    "request_id": request_id,
                }
            }, 400
        except PermissionError as e:
            # 权限错误
            return {
                "error": {
                    "code": "FORBIDDEN",
                    "message": str(e),
                }
            }, 403
        except LookupError as e:
            # 资源不存在（KeyError, IndexError 等）
            return {
                "error": {
                    "code": "NOT_FOUND",
                    "message": str(e),
                }
            }, 404
        except Exception as e:
            # 未知错误
            request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
            logger.exception(
                "服务器错误 [%s]: %s %s",
                request_id,
                request.method,
                request.path,
            )
            return {
                "error": {
                    "code": "INTERNAL_ERROR",
                    "message": "服务器内部错误",
                    "request_id": request_id,
                }
            }, 500

    return wrapper


@app.route("/articles/<int:article_id>")
@handle_api_errors
def get_article(article_id: int) -> dict[str, Any]:
    """获取文章 - 自动处理异常"""
    articles: dict[int, dict[str, str]] = {
        1: {"id": "1", "title": "Python 入门"},
    }
    # KeyError 会被装饰器捕获并转换为 404
    return articles[article_id]


@app.route("/articles", methods=["POST"])
@handle_api_errors
def create_article() -> tuple[dict[str, str], int]:
    """创建文章 - 自动处理验证错误"""
    data: dict[str, Any] | None = request.get_json()
    if data is None or "title" not in data:
        raise ValueError("标题是必填字段")
    return {"message": "创建成功"}, 201

2.3 不同错误类型的错误码设计

业务错误码分类体系：

错误码设计规范：
┌──────────────────────────────────────────────────────────┐
│  格式：模块_错误类型                                      │
│                                                          │
│  用户模块：                                               │
│  ├── USER_NOT_FOUND           用户不存在                  │
│  ├── USER_ALREADY_EXISTS      用户已存在                  │
│  ├── USER_INVALID_PASSWORD    密码不正确                  │
│  └── USER_ACCOUNT_DISABLED    账号已禁用                  │
│                                                          │
│  文章模块：                                               │
│  ├── ARTICLE_NOT_FOUND        文章不存在                  │
│  ├── ARTICLE_PERMISSION_DENIED 无权操作此文章             │
│  └── ARTICLE_PUBLISH_FAILED   发布失败                    │
│                                                          │
│  通用模块：                                               │
│  ├── VALIDATION_ERROR         参数验证失败                │
│  ├── AUTHENTICATION_REQUIRED  需要认证                   │
│  ├── FORBIDDEN                无权访问                    │
│  ├── RATE_LIMIT_EXCEEDED      请求频率过高                │
│  └── INTERNAL_ERROR           服务器内部错误              │
└──────────────────────────────────────────────────────────┘

python

# error_codes.py
from enum import Enum
from typing import Any


class ErrorCode(Enum):
    """业务错误码枚举"""

    # 用户相关
    USER_NOT_FOUND = ("USER_NOT_FOUND", "用户不存在", 404)
    USER_ALREADY_EXISTS = ("USER_ALREADY_EXISTS", "用户已存在", 409)
    USER_INVALID_PASSWORD = ("USER_INVALID_PASSWORD", "密码不正确", 401)
    USER_ACCOUNT_DISABLED = ("USER_ACCOUNT_DISABLED", "账号已禁用", 403)

    # 文章相关
    ARTICLE_NOT_FOUND = ("ARTICLE_NOT_FOUND", "文章不存在", 404)
    ARTICLE_PERMISSION_DENIED = (
        "ARTICLE_PERMISSION_DENIED",
        "无权操作此文章",
        403,
    )

    # 通用
    VALIDATION_ERROR = ("VALIDATION_ERROR", "参数验证失败", 400)
    AUTHENTICATION_REQUIRED = ("AUTHENTICATION_REQUIRED", "需要认证", 401)
    FORBIDDEN = ("FORBIDDEN", "无权访问", 403)
    RATE_LIMIT_EXCEEDED = ("RATE_LIMIT_EXCEEDED", "请求频率过高", 429)
    INTERNAL_ERROR = ("INTERNAL_ERROR", "服务器内部错误", 500)

    @property
    def code(self) -> str:
        return self.value[0]

    @property
    def message(self) -> str:
        return self.value[1]

    @property
    def status_code(self) -> int:
        return self.value[2]


class ApiError(Exception):
    """API 业务异常类"""

    def __init__(
        self,
        error_code: ErrorCode,
        message: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        self.error_code = error_code
        self.message = message or error_code.message
        self.details = details
        super().__init__(self.message)

    def to_response(self) -> tuple[dict[str, Any], int]:
        """转换为 HTTP 响应"""
        response: dict[str, Any] = {
            "error": {
                "code": self.error_code.code,
                "message": self.message,
            }
        }
        if self.details is not None:
            response["error"]["details"] = self.details
        return response, self.error_code.status_code


# 使用示例
def get_user_service(user_id: int) -> dict[str, Any]:
    """获取用户 - 抛出业务异常"""
    users: dict[int, dict[str, str]] = {1: {"name": "Alice"}}
    user: dict[str, str] | None = users.get(user_id)
    if user is None:
        raise ApiError(ErrorCode.USER_NOT_FOUND, details={"user_id": user_id})
    return user


@app.errorhandler(ApiError)
def handle_api_error(error: ApiError) -> tuple[dict[str, Any], int]:
    """统一处理 ApiError"""
    return error.to_response()

第三部分：日志（L1）

3.1 app.logger 基本用法

实际场景：

应用运行时需要记录关键信息用于排查问题，如用户登录、接口调用、错误堆栈等。

问题：Flask 内置的日志如何使用？

python

# logging_basics.py
from flask import Flask, request
from typing import Any
import logging

app: Flask = Flask(__name__)


@app.route("/users/<int:user_id>")
def get_user(user_id: int) -> dict[str, str]:
    """获取用户 - 记录各级别日志"""
    users: dict[int, dict[str, str]] = {1: {"name": "Alice"}}

    # DEBUG - 调试信息（开发时查看详细流程）
    app.logger.debug("查询用户: user_id=%d, method=%s", user_id, request.method)

    # INFO - 关键业务信息（记录正常业务流程）
    app.logger.info("用户查询成功: user_id=%d", user_id)

    # WARNING - 警告信息（异常但系统仍正常运行）
    if user_id > 1000:
        app.logger.warning("查询了不存在的用户 ID: user_id=%d", user_id)

    user: dict[str, str] | None = users.get(user_id)
    if user is None:
        # ERROR - 错误信息（功能异常或请求失败）
        app.logger.error("用户不存在: user_id=%d", user_id)
        from flask import abort

        abort(404)

    return user

日志级别说明：

日志级别（从低到高）：
┌──────────────────────────────────────────────────────────┐
│  级别          数值    用途                               │
│  ─────────────────────────────────────────────────────   │
│  DEBUG         10     调试信息，开发环境使用               │
│  INFO          20     关键业务操作（登录、下单等）         │
│  WARNING       30     警告（配置缺失、降级运行等）         │
│  ERROR         40     错误（请求失败、数据库连接断开等）   │
│  CRITICAL      50     严重错误（系统即将崩溃）             │
│                                                          │
│  设置级别后，只记录 >= 该级别的信息                       │
│  app.logger.setLevel(logging.DEBUG) → 记录所有            │
│  app.logger.setLevel(logging.INFO)  → 记录 INFO 及以上    │
│  app.logger.setLevel(logging.WARNING) → 记录 WARNING 及以上│
└──────────────────────────────────────────────────────────┘

3.2 日志配置：格式、输出目标

实际场景：

默认的日志格式不包含时间戳，而且输出到 stderr 不方便排查生产问题。

问题：如何自定义日志格式和输出目标？

python

# logging_config.py
from flask import Flask
import logging
import os
from logging.handlers import RotatingFileHandler


def setup_logging(app: Flask) -> None:
    """配置 Flask 应用日志"""

    # 日志格式
    log_format: str = (
        "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s"
    )
    formatter: logging.Formatter = logging.Formatter(log_format)

    # 控制台输出
    console_handler: logging.StreamHandler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    app.logger.addHandler(console_handler)

    # 文件输出（按大小轮转）
    log_dir: str = os.path.join(os.path.dirname(__file__), "logs")
    os.makedirs(log_dir, exist_ok=True)
    file_handler: RotatingFileHandler = RotatingFileHandler(
        os.path.join(log_dir, "app.log"),
        maxBytes=10 * 1024 * 1024,  # 10MB
        backupCount=5,
    )
    file_handler.setFormatter(formatter)
    app.logger.addHandler(file_handler)

    # 设置日志级别
    if app.debug:
        app.logger.setLevel(logging.DEBUG)
    else:
        app.logger.setLevel(logging.INFO)


app: Flask = Flask(__name__)
setup_logging(app)


@app.route("/")
def index() -> str:
    app.logger.info("访问首页")
    return "Hello"

日志输出示例：

2024-01-15 10:30:45,123 [INFO] article_service:25 - 用户查询成功: user_id=1
2024-01-15 10:30:46,456 [WARNING] article_service:30 - 查询了不存在的用户 ID: user_id=9999
2024-01-15 10:30:47,789 [ERROR] article_service:35 - 用户不存在: user_id=9999

字段	含义
`%(asctime)s`	时间戳
`%(levelname)s`	日志级别（DEBUG/INFO/ERROR）
`%(name)s`	日志器名称（通常是模块名）
`%(lineno)d`	代码行号
`%(message)s`	日志消息

第四部分：L2 实践层

4.1 邮件告警（生产环境错误通知）

实际场景：

生产环境发生 500 错误时，开发团队需要第一时间收到通知，而不是等用户反馈。

问题：如何在错误发生时自动发送邮件告警？

python

# email_alert.py
from flask import Flask
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from logging.handlers import SMTPHandler
from typing import Any


class EmailAlertHandler(logging.Handler):
    """自定义邮件告警处理器"""

    def __init__(
        self,
        smtp_host: str,
        smtp_port: int,
        from_addr: str,
        to_addrs: list[str],
        username: str,
        password: str,
        subject_prefix: str = "[Flask Alert]",
    ) -> None:
        super().__init__(logging.ERROR)
        self.smtp_host = smtp_host
        self.smtp_port = smtp_port
        self.from_addr = from_addr
        self.to_addrs = to_addrs
        self.username = username
        self.password = password
        self.subject_prefix = subject_prefix

    def emit(self, record: logging.LogRecord) -> None:
        try:
            subject: str = f"{self.subject_prefix} {record.levelname}: {record.getMessage()}"
            body: str = self.format(record)

            msg: MIMEMultipart = MIMEMultipart()
            msg["Subject"] = subject
            msg["From"] = self.from_addr
            msg["To"] = ", ".join(self.to_addrs)
            msg.attach(MIMEText(body, "plain", "utf-8"))

            with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
                server.starttls()
                server.login(self.username, self.password)
                server.sendmail(self.from_addr, self.to_addrs, msg.as_string())
        except Exception:
            self.handleError(record)


def setup_email_alert(app: Flask) -> None:
    """配置邮件告警"""
    handler: EmailAlertHandler = EmailAlertHandler(
        smtp_host="smtp.example.com",
        smtp_port=587,
        from_addr="alert@example.com",
        to_addrs=["dev-team@example.com"],
        username="alert@example.com",
        password="your_password",
        subject_prefix="[生产环境 Flask]",
    )

    # 设置邮件日志格式（包含完整堆栈）
    mail_formatter: logging.Formatter = logging.Formatter(
        "%(asctime)s\n"
        "级别: %(levelname)s\n"
        "模块: %(name)s\n"
        "行号: %(lineno)d\n"
        "消息: %(message)s\n"
        "\n--- 完整堆栈 ---\n"
        "%(exc_text)s"
    )
    handler.setFormatter(mail_formatter)

    # 只处理 ERROR 及以上级别
    handler.setLevel(logging.ERROR)
    app.logger.addHandler(handler)


app: Flask = Flask(__name__)
if not app.debug:
    setup_email_alert(app)

使用内置 SMTPHandler（简化版）：

python

# simple_email_alert.py
from flask import Flask
from logging.handlers import SMTPHandler
import logging


app: Flask = Flask(__name__)

if not app.debug:
    mail_handler: SMTPHandler = SMTPHandler(
        mailhost=("smtp.example.com", 587),
        fromaddr="alert@example.com",
        toaddrs=["dev@example.com"],
        subject="[生产环境] Flask 应用错误",
        credentials=("alert@example.com", "your_password"),
    )
    mail_handler.setLevel(logging.ERROR)
    mail_handler.setFormatter(
        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    )
    app.logger.addHandler(mail_handler)

4.2 请求信息注入日志

实际场景：

日志中只有错误信息，但无法知道是哪个请求、哪个用户触发的，排查困难。

问题：如何在每条日志中自动包含请求上下文？

python

# request_logging.py
from flask import Flask, request, g
from typing import Any
import logging
import uuid


class RequestFormatter(logging.Formatter):
    """带请求信息的日志格式化器"""

    def format(self, record: logging.LogRecord) -> str:
        # 从 g 对象获取请求信息
        record.request_id = getattr(g, "request_id", "N/A")
        record.request_method = getattr(g, "request_method", "N/A")
        record.request_url = getattr(g, "request_url", "N/A")
        record.remote_addr = getattr(g, "remote_addr", "N/A")
        record.user_agent = getattr(g, "user_agent", "N/A")
        return super().format(record)


def setup_request_logging(app: Flask) -> None:
    """配置请求信息注入日志"""

    @app.before_request
    def inject_request_info() -> None:
        """在每个请求开始前注入上下文"""
        g.request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
        g.request_method = request.method
        g.request_url = request.url
        g.remote_addr = request.remote_addr
        g.user_agent = request.headers.get("User-Agent", "Unknown")

    # 配置日志格式化器
    log_format: str = (
        "%(asctime)s [%(levelname)s] [%(request_id)s] "
        "%(request_method)s %(request_url)s %(remote_addr)s - "
        "%(message)s"
    )
    formatter: RequestFormatter = RequestFormatter(log_format)

    # 应用到所有 handler
    for handler in app.logger.handlers:
        handler.setFormatter(formatter)


app: Flask = Flask(__name__)
setup_request_logging(app)


@app.route("/articles/<int:article_id>")
def get_article(article_id: int) -> dict[str, Any]:
    articles: dict[int, dict[str, str]] = {1: {"title": "Python 入门"}}
    article: dict[str, str] | None = articles.get(article_id)
    if article is None:
        # 日志自动包含 request_id, method, url, IP
        app.logger.error("文章不存在: article_id=%d", article_id)
        from flask import abort

        abort(404)
    return article

日志输出效果：

2024-01-15 10:30:45,123 [ERROR] [a1b2c3d4] GET /articles/999 192.168.1.100 - 文章不存在: article_id=999
2024-01-15 10:30:46,456 [INFO] [e5f6g7h8] POST /articles 192.168.1.101 - 文章创建成功

4.3 错误页模板设计

实际场景：

默认的 404/500 页面是纯文本的，用户体验差且不专业。

问题：如何设计美观的错误页面模板？

错误页模板目录结构：

templates/
├── errors/
│   ├── 400.html    请求错误
│   ├── 403.html    禁止访问
│   ├── 404.html    页面不存在
│   ├── 500.html    服务器错误
│   └── base.html   错误页基础模板

错误页基础模板：

html

<!-- templates/errors/base.html -->
<!DOCTYPE html>
<html lang="zh-CN">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>{% block title %}错误{% endblock %}</title>
    <style>
      body {
        font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
        display: flex;
        align-items: center;
        justify-content: center;
        min-height: 100vh;
        margin: 0;
        background: #f5f5f5;
      }
      .error-container {
        text-align: center;
        padding: 40px;
        max-width: 500px;
      }
      .error-code {
        font-size: 72px;
        font-weight: bold;
        color: #333;
        margin: 0;
      }
      .error-message {
        font-size: 18px;
        color: #666;
        margin: 16px 0 32px;
      }
      .back-link {
        display: inline-block;
        padding: 12px 24px;
        background: #007bff;
        color: white;
        text-decoration: none;
        border-radius: 4px;
      }
      .back-link:hover {
        background: #0056b3;
      }
    </style>
  </head>
  <body>
    <div class="error-container">
      {% block content %}{% endblock %}
    </div>
  </body>
</html>

404 错误页：

html

<!-- templates/errors/404.html -->
{% extends "errors/base.html" %} {% block title %}页面未找到{% endblock %} {%
block content %}
<p class="error-code">404</p>
<p class="error-message">抱歉，您访问的页面不存在</p>
<a href="/" class="back-link">返回首页</a>
{% endblock %}

500 错误页：

html

<!-- templates/errors/500.html -->
{% extends "errors/base.html" %} {% block title %}服务器错误{% endblock %} {%
block content %}
<p class="error-code">500</p>
<p class="error-message">服务器开小差了，请稍后重试</p>
<a href="/" class="back-link">返回首页</a>
{% endblock %}

4.4 最佳实践表格

错误处理最佳实践：

做法	原因	示例
捕获具体异常	避免隐藏未知错误	`except ValueError:` 而非 `except:`
使用 `abort()` 快速失败	代码清晰，语义明确	`if not user: abort(404)`
统一错误响应格式	前端易于解析处理	`{ "error": { "code", "message" } }`
记录完整上下文	方便排查问题	包含 request_id、URL、IP
生产环境关闭调试	避免泄露敏感信息	`app.debug = False`
区分客户端/服务端错误	4xx vs 5xx 处理策略不同	4xx 返回提示，5xx 告警

日志最佳实践：

做法	原因	示例
结构化日志格式	便于日志聚合工具解析	JSON 格式或固定格式
日志级别合理分级	DEBUG（开发）/INFO（业务）/ERROR（告警）	不同级别不同处理
日志轮转	防止磁盘写满	`RotatingFileHandler`
敏感信息脱敏	保护用户隐私	密码、token 不记录明文
包含请求追踪 ID	串联请求全链路	`X-Request-ID`

4.5 反模式

反模式 1：裸 except 吞掉所有异常

python

# ❌ 错误做法 - 裸 except
@app.route("/users/<int:user_id>")
def get_user_bad(user_id: int) -> dict[str, str]:
    try:
        user: dict[str, str] = database.get(user_id)
        return user
    except:  # 裸 except 吞掉所有异常，包括 KeyboardInterrupt
        return {}

python

# ✅ 正确做法 - 捕获具体异常
@app.route("/users/<int:user_id>")
def get_user_good(user_id: int) -> dict[str, str]:
    try:
        user: dict[str, str] = database.get(user_id)
        return user
    except KeyError:
        app.logger.warning("用户不存在: user_id=%d", user_id)
        from flask import abort

        abort(404)
    except ConnectionError as e:
        app.logger.error("数据库连接失败: %s", str(e))
        abort(503)

反模式 2：吞掉异常不处理

python

# ❌ 错误做法 - 捕获后不处理
@app.route("/process")
def process_bad() -> str:
    try:
        result: str = risky_operation()
        return result
    except Exception:
        pass  # 什么都没做，错误被静默吞掉

python

# ✅ 正确做法 - 记录并处理
@app.route("/process")
def process_good() -> str:
    try:
        result: str = risky_operation()
        return result
    except ValueError as e:
        app.logger.error("处理失败: %s", str(e))
        return "处理失败，请检查输入参数"
    except Exception as e:
        app.logger.exception("未知错误")  # exception 自动包含堆栈
        from flask import abort

        abort(500)

反模式 3：生产环境 debug=True

python

# ❌ 错误做法 - 生产环境开启调试
app: Flask = Flask(__name__)
app.run(debug=True, host="0.0.0.0")  # 危险！暴露交互式调试器

python

# ✅ 正确做法 - 根据环境配置
# config.py
import os


class Config:
    DEBUG: bool = False
    TESTING: bool = False


class DevelopmentConfig(Config):
    DEBUG: bool = True


class ProductionConfig(Config):
    DEBUG: bool = False
    LOG_LEVEL: str = "INFO"


class TestingConfig(Config):
    TESTING: bool = True


# main.py
from flask import Flask


def create_app() -> Flask:
    env: str = os.getenv("FLASK_ENV", "production")
    app: Flask = Flask(__name__)

    if env == "development":
        app.config.from_object(DevelopmentConfig)
    elif env == "testing":
        app.config.from_object(TestingConfig)
    else:
        app.config.from_object(ProductionConfig)

    return app

反模式 4：错误响应格式不一致

python

# ❌ 错误做法 - 格式混乱
@app.route("/articles/<int:article_id>")
def get_article_inconsistent(article_id: int) -> Any:
    if article_id < 0:
        return "文章 ID 无效", 400  # 纯字符串
    article: dict[str, Any] | None = db.get(article_id)
    if article is None:
        return {"error": "not found"}  # JSON 但结构不同
    return article

python

# ✅ 正确做法 - 统一格式
@app.route("/articles/<int:article_id>")
def get_article_consistent(article_id: int) -> tuple[Any, int]:
    if article_id < 0:
        return make_api_error("INVALID_ID", "文章 ID 无效", 400)
    article: dict[str, Any] | None = db.get(article_id)
    if article is None:
        return make_api_error("ARTICLE_NOT_FOUND", "文章不存在", 404)
    return article, 200

第五部分：L3 专家层

5.1 Flask 错误处理流程

完整错误处理链路：

Flask 错误处理流程：
┌──────────────────────────────────────────────────────────────────┐
│                                                                  │
│  请求到达                                                        │
│       │                                                          │
│       ↓                                                          │
│  ┌──────────────┐                                                │
│  │  路由匹配     │                                                │
│  └──────────────┘                                                │
│       │                                                          │
│       ↓                                                          │
│  ┌──────────────┐                                                │
│  │  视图函数执行  │                                                │
│  └──────────────┘                                                │
│       │                                                          │
│       ├────────────────────┐                                     │
│       │                    │                                     │
│       ↓                    ↓                                     │
│  正常返回             抛出异常                                    │
│       │                    │                                     │
│       │                    ↓                                     │
│       │             ┌──────────────┐                             │
│       │             │ abort() 调用 │                             │
│       │             │ 或异常抛出   │                             │
│       │             └──────────────┘                             │
│       │                    │                                     │
│       │                    ↓                                     │
│       │             ┌──────────────────────────────┐             │
│       │             │  查找错误处理器               │             │
│       │             │  1. Blueprint errorhandler   │             │
│       │             │  2. Application errorhandler │             │
│       │             │  3. Flask 默认处理           │             │
│       │             └──────────────────────────────┘             │
│       │                    │                                     │
│       │                    ↓                                     │
│       │             ┌──────────────┐                             │
│       │             │  执行处理逻辑 │                             │
│       │             └──────────────┘                             │
│       │                    │                                     │
│       ↓                    ↓                                     │
│  ┌──────────────────────────────┐                                │
│  │         构建响应              │                                │
│  └──────────────────────────────┘                                │
│       │                                                          │
│       ↓                                                          │
│  响应返回客户端                                                   │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

python

# error_flow_demo.py
from flask import Flask, abort, jsonify, request
from werkzeug.exceptions import HTTPException
from typing import Any

app: Flask = Flask(__name__)


@app.route("/demo/<int:item_id>")
def demo_error_flow(item_id: int) -> dict[str, Any]:
    """演示错误处理流程"""
    app.logger.debug("开始处理请求: item_id=%d", item_id)

    # 场景 1: 主动调用 abort
    if item_id == 0:
        abort(400, description="ID 不能为 0")

    # 场景 2: 业务异常
    items: dict[int, str] = {1: "item1", 2: "item2"}
    if item_id not in items:
        raise KeyError(f"物品 {item_id} 不存在")

    return {"item": items[item_id]}


# 注册错误处理器（在路由之后注册也可以，Flask 会正确匹配）
@app.errorhandler(HTTPException)
def handle_http_exception(error: HTTPException) -> tuple[dict[str, Any], int]:
    """统一处理所有 HTTP 异常"""
    response: dict[str, Any] = {
        "error": {
            "code": error.code,
            "message": error.description or str(error),
        }
    }
    return response, error.code if error.code else 500


@app.errorhandler(KeyError)
def handle_key_error(error: KeyError) -> tuple[dict[str, Any], int]:
    """处理 KeyError 转换为 404"""
    return {"error": {"code": 404, "message": str(error)}}, 404

5.2 Werkzeug 调试器原理

Development Mode 下的交互式调试：

Werkzeug 调试器工作流程：
┌──────────────────────────────────────────────────────────┐
│                                                          │
│  debug=True 时启动                                       │
│       │                                                  │
│       ↓                                                  │
│  ┌──────────────────────────┐                            │
│  │  Werkzeug Debugger 接管   │                            │
│  │  - 捕获未处理异常         │                            │
│  │  - 生成交互式调试页面     │                            │
│  │  - 提供 PIN 码保护        │                            │
│  └──────────────────────────┘                            │
│       │                                                  │
│       ↓                                                  │
│  ┌──────────────────────────┐                            │
│  │  调试页面功能             │                            │
│  │  1. 显示完整 Traceback    │                            │
│  │  2. 查看每个栈帧变量      │                            │
│  │  3. 交互式 Python 控制台 │                            │
│  │     （需要 PIN 码）       │                            │
│  └──────────────────────────┘                            │
│                                                          │
│  ⚠️  安全风险：生产环境绝不可开启                          │
│     - 攻击者可执行任意 Python 代码                        │
│     - PIN 码可能被暴力破解                                │
│                                                          │
└──────────────────────────────────────────────────────────┘

python

# debugger_demo.py
from flask import Flask

app: Flask = Flask(__name__)


@app.route("/bug")
def trigger_bug() -> str:
    """触发错误查看调试器"""
    data: dict[str, str] = {"name": "Alice"}
    # 故意访问不存在的键
    return data["nonexistent_key"]  # 触发 KeyError


if __name__ == "__main__":
    # debug=True 时，Werkzeug 提供交互式调试
    # 访问 /bug 会看到：
    # 1. 完整的 Traceback（带代码高亮）
    # 2. 每个栈帧可点击展开查看局部变量
    # 3. 控制台图标 → 输入 PIN 码 → 交互式 Python shell
    app.run(debug=True)

调试器 PIN 码机制：

PIN 码生成原理：
┌──────────────────────────────────────────────────────────┐
│  PIN 码基于以下因素计算：                                  │
│  1. 当前用户名                                             │
│  2. 模块名称 (flask.app)                                  │
│  3. Flask 安装路径                                         │
│  4. 机器 ID（网络接口 MAC 地址或机器标识）                  │
│                                                          │
│  首次启动时打印：                                          │
│  " * Debugger is active!"                                │
│  " * Debugger PIN: 123-456-789"                          │
│                                                          │
│  保护机制：                                                │
│  - PIN 错误超过限制后需要等待                              │
│  - PIN 码每次启动可能变化                                  │
│  - 只有本地访问时才展示完整调试功能                        │
└──────────────────────────────────────────────────────────┘

5.3 日志聚合方案

生产环境日志架构：

日志聚合架构：
┌──────────────────────────────────────────────────────────────────┐
│                                                                  │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐                 │
│  │  Flask   │────→│  File/   │────→│  Log     │                 │
│  │  App     │     │  Stdout  │     │  Shipper │                 │
│  └──────────┘     └──────────┘     │ (Filebeat│                 │
│                                    │  /Fluentd)│                 │
│                                    └──────────┘                 │
│                                          │                       │
│                                          ↓                       │
│                                    ┌──────────┐                 │
│                                    │  Message │                 │
│                                    │  Queue   │                 │
│                                    │ (Redis/  │                 │
│                                    │  Kafka)  │                 │
│                                    └──────────┘                 │
│                                          │                       │
│                                          ↓                       │
│                                    ┌──────────┐                 │
│                                    │  Log     │                 │
│                                    │  Engine  │                 │
│                                    │(Elastic- │                 │
│                                    │  search) │                 │
│                                    └──────────┘                 │
│                                          │                       │
│                                          ↓                       │
│                                    ┌──────────┐                 │
│                                    │  Kibana  │                 │
│                                    │  (可视化  │                 │
│                                    │   搜索)  │                 │
│                                    └──────────┘                 │
│                                                                  │
│  或: Flask ─→ Sentry SDK ─→ Sentry Server ─→ Web 告警           │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

方案对比：

方案	优点	缺点	适用场景
ELK Stack	功能强大，灵活度高	部署复杂，资源占用大	大型企业，完整日志平台
Sentry	开箱即用，错误追踪强	付费版有事件限制	中小团队，错误监控
Loki + Grafana	轻量，与 Grafana 集成好	全文搜索能力弱	已有 Grafana 的团队
云厂商日志服务	免运维，按量计费	厂商锁定	使用对应云服务的团队

集成 Sentry 示例：

python

# sentry_integration.py
from flask import Flask, request
from typing import Any
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration


def init_sentry(dsn: str, environment: str = "production") -> None:
    """初始化 Sentry SDK"""
    sentry_sdk.init(
        dsn=dsn,
        integrations=[FlaskIntegration()],
        environment=environment,
        # 采样率（生产环境可适当降低）
        traces_sample_rate=1.0,
        # 添加自定义标签
        before_send=lambda event, hint: _enrich_sentry_event(event, hint),
    )


def _enrich_sentry_event(event: dict[str, Any], hint: Any) -> dict[str, Any]:
    """在发送前丰富 Sentry 事件"""
    from flask import has_request_context, g

    if has_request_context():
        # 添加 request_id
        event.setdefault("tags", {})["request_id"] = getattr(g, "request_id", "N/A")
    return event


app: Flask = Flask(__name__)
init_sentry(
    dsn="https://your_sentry_dsn@sentry.io/123456",
    environment="production",
)


@app.route("/checkout")
def checkout() -> dict[str, str]:
    """业务逻辑 - Sentry 自动捕获未处理异常"""
    # 手动捕获异常（带额外上下文）
    try:
        process_payment()
    except PaymentError as e:
        sentry_sdk.capture_exception(e)
        return {"error": "支付失败"}, 500
    return {"status": "success"}


def process_payment() -> None:
    raise PaymentError("余额不足")


class PaymentError(Exception):
    pass

5.4 知识关联图

错误处理与日志知识关联：
┌─────────────────────────────────────────────────────────────────┐
│                                                                 │
│                    ┌─────────────────────┐                      │
│                    │   Python 异常机制    │                      │
│                    │   try/except/raise  │                      │
│                    └─────────────────────┘                      │
│                              │                                  │
│                    ┌─────────┴─────────┐                        │
│                    ↓                   ↓                        │
│           ┌──────────────┐    ┌──────────────┐                 │
│           │  HTTP 错误    │    │  自定义异常   │                 │
│           │  abort(4xx)  │    │  ApiError    │                 │
│           └──────────────┘    └──────────────┘                 │
│                    │                   │                        │
│                    └─────────┬─────────┘                        │
│                              ↓                                  │
│                    ┌─────────────────────┐                      │
│                    │  @app.errorhandler  │                      │
│                    │  统一错误处理        │                      │
│                    └─────────────────────┘                      │
│                              │                                  │
│              ┌───────────────┼───────────────┐                  │
│              ↓               ↓               ↓                  │
│     ┌──────────────┐ ┌──────────────┐ ┌──────────────┐         │
│     │  HTML 错误页  │ │  JSON 错误   │ │  日志记录    │         │
│     │  模板渲染     │ │  统一格式    │ │  app.logger  │         │
│     └──────────────┘ └──────────────┘ └──────────────┘         │
│              │               │               │                  │
│              ↓               ↓               ↓                  │
│     ┌──────────────┐ ┌──────────────┐ ┌──────────────┐         │
│     │  用户体验     │ │  前端解析    │ │  日志聚合    │         │
│     │  友好提示     │ │  错误码      │ │  ELK/Sentry │         │
│     └──────────────┘ └──────────────┘ └──────────────┘         │
│                                                                 │
│  前置知识：Python 异常处理、HTTP 协议                           │
│  进阶扩展：分布式链路追踪、APM 监控、告警策略                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

本章总结

错误处理与日志核心要点：
┌──────────────────────────────────────────────────────────┐
│                                                          │
│  L1 会用                                                  │
│  ├── abort() 抛出 HTTP 错误                              │
│  ├── @app.errorhandler() 自定义错误处理器                │
│  ├── Blueprint 级别错误处理器                             │
│  ├── 统一 JSON 错误响应格式                               │
│  └── app.logger 记录各级别日志                            │
│                                                          │
│  L2 用好                                                  │
│  ├── 邮件告警通知生产错误                                 │
│  ├── 请求信息注入日志（request_id, URL, IP）             │
│  ├── 错误页模板设计                                       │
│  ├── 避免裸 except / 吞异常 / 生产 debug                 │
│  └── 统一错误码设计（ErrorCode 枚举）                    │
│                                                          │
│  L3 深入                                                  │
│  ├── Flask 错误处理完整流程                              │
│  ├── Werkzeug 调试器原理与安全风险                        │
│  ├── 日志聚合方案（ELK / Sentry / Loki）                 │
│  └── 错误追踪与分布式链路                                 │
│                                                          │
└──────────────────────────────────────────────────────────┘

核心原则：

快速失败：用 abort() 在错误点立即中断，而非继续执行
统一格式：API 错误响应保持一致的结构，便于前端处理
完整上下文：日志包含 request_id、URL、IP 等排查关键信息
分级告警：ERROR 级别触发邮件/钉钉通知，不要所有日志都告警
生产安全：debug=False，关闭 Werkzeug 调试器，避免代码泄露

13-错误处理与日志 ​

第一部分：错误处理（L1） ​

1.1 HTTP 错误：abort() 函数 ​

1.2 自定义错误处理器 ​

1.3 错误处理器的返回值 ​

1.4 Blueprint 级别的错误处理器 ​

第二部分：JSON API 错误（L1） ​

2.1 RESTful API 标准错误响应格式 ​

2.2 统一错误处理装饰器 ​

2.3 不同错误类型的错误码设计 ​

第三部分：日志（L1） ​

3.1 app.logger 基本用法 ​

3.2 日志配置：格式、输出目标 ​

第四部分：L2 实践层 ​

4.1 邮件告警（生产环境错误通知） ​

4.2 请求信息注入日志 ​

4.3 错误页模板设计 ​

4.4 最佳实践表格 ​

4.5 反模式 ​

反模式 1：裸 except 吞掉所有异常 ​

反模式 2：吞掉异常不处理 ​

反模式 3：生产环境 debug=True ​

反模式 4：错误响应格式不一致 ​

第五部分：L3 专家层 ​

5.1 Flask 错误处理流程 ​

5.2 Werkzeug 调试器原理 ​

5.3 日志聚合方案 ​

5.4 知识关联图 ​

本章总结 ​

13-错误处理与日志

第一部分：错误处理（L1）

1.1 HTTP 错误：abort() 函数

1.2 自定义错误处理器

1.3 错误处理器的返回值

1.4 Blueprint 级别的错误处理器

第二部分：JSON API 错误（L1）

2.1 RESTful API 标准错误响应格式

2.2 统一错误处理装饰器

2.3 不同错误类型的错误码设计

第三部分：日志（L1）

3.1 app.logger 基本用法

3.2 日志配置：格式、输出目标

第四部分：L2 实践层

4.1 邮件告警（生产环境错误通知）

4.2 请求信息注入日志

4.3 错误页模板设计

4.4 最佳实践表格

4.5 反模式

反模式 1：裸 except 吞掉所有异常

反模式 2：吞掉异常不处理

反模式 3：生产环境 debug=True

反模式 4：错误响应格式不一致

第五部分：L3 专家层

5.1 Flask 错误处理流程

5.2 Werkzeug 调试器原理

5.3 日志聚合方案

5.4 知识关联图

本章总结