中间件、异常处理与自定义响应

💡 本部分目标：学会捕获错误并返回友好提示；使用中间件记录日志、处理跨域（CORS）；统一 API 响应格式，提升应用健壮性和用户体验。

---

一、为什么需要异常处理和中间件？

• 用户友好：默认错误信息（如 422）对前端不友好
• 安全性：避免暴露内部错误细节
• 可维护性：统一日志、性能监控、跨域支持
• 一致性：所有接口返回相同结构的 JSON

FastAPI 提供了强大的机制来实现这些需求。

---

二、自定义异常处理器（Exception Handlers）

2.1 捕获内置异常

FastAPI 自动处理很多异常（如验证失败），但你可以覆盖它。

示例：美化 422 错误（请求数据无效）

# main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=400,
        content={
            "error": "请求参数错误",
            "details": exc.errors()
        }
    )

🔍 现在，当 JSON 字段缺失或类型错误时，返回：

{
  "error": "请求参数错误",
  "details": [ ... ]
}

2.2 捕获自定义异常

步骤1：定义自定义异常类

# exceptions.py
class UserNotFoundException(Exception):
    def __init__(self, username: str):
        self.username = username

步骤2：注册异常处理器

# main.py
from exceptions import UserNotFoundException

@app.exception_handler(UserNotFoundException)
async def user_not_found_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"error": f"用户 {exc.username} 不存在"}
    )

步骤3：在路由中抛出异常

@app.get("/users/{username}")
def get_user(username: str):
    if username not in ["alice", "bob"]:
        raise UserNotFoundException(username)
    return {"username": username}

---

三、中间件（Middleware）

中间件在每个请求前后执行通用逻辑，如：

• 记录请求日志
• 添加 CORS 头（允许前端跨域）
• 测量请求耗时
• 验证全局 Token

3.1 CORS 中间件（必须！）

前端（如 React/Vue）通常运行在不同端口，浏览器会阻止跨域请求。
CORS（Cross-Origin Resource Sharing） 解决这个问题。

# main.py
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许所有来源（生产环境应限制）
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有 HTTP 方法
    allow_headers=["*"],  # 允许所有请求头
)

⚠️ 生产环境建议：

allow_origins=["https://your-frontend.com"]

3.2 自定义中间件：记录请求日志

import time
from starlette.middleware.base import BaseHTTPMiddleware

class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        print(f"请求: {request.method} {request.url} | 耗时: {process_time:.3f}s")
        return response

app.add_middleware(LogMiddleware)

🔍 启动后，每次请求会打印日志到控制台。

---

四、统一响应格式（最佳实践）

很多团队要求所有成功响应都遵循固定结构，例如：

{
  "code": 200,
  "message": "success",
  "data": { ... }
}

实现方式：自定义响应模型 + 封装函数

# schemas.py
from pydantic import BaseModel
from typing import Any, Optional

class ResponseModel(BaseModel):
    code: int = 200
    message: str = "success"
    data: Any = None

# 工具函数
def success_response(data: Any = None, message: str = "success"):
    return ResponseModel(data=data, message=message)

在路由中使用

@app.get("/health")
def health_check():
    return success_response(data={"status": "OK"})

💡 你也可以通过中间件自动包装所有响应，但初学者建议先用函数封装。

---

五、完整示例代码（main.py）

# main.py
import time
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from pydantic import BaseModel
from typing import Any

# 自定义异常
class UserNotFoundException(Exception):
    def __init__(self, username: str):
        self.username = username

# 统一响应模型
class ResponseModel(BaseModel):
    code: int = 200
    message: str = "success"
    data: Any = None

def success_response(data: Any = None, message: str = "success"):
    return ResponseModel(data=data, message=message)

# 日志中间件
class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        print(f"⏱️ {request.method} {request.url} | {process_time:.3f}s")
        return response

# 创建应用
app = FastAPI(title="第7部分：异常处理与中间件")

# 添加中间件（顺序很重要！）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(LogMiddleware)

# 异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=400,
        content={"code": 400, "message": "请求参数错误", "data": exc.errors()}
    )

@app.exception_handler(UserNotFoundException)
async def user_not_found_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"code": 404, "message": f"用户 {exc.username} 不存在", "data": None}
    )

# 路由
@app.get("/health")
def health_check():
    return success_response(data={"status": "OK"})

@app.get("/users/{username}")
def get_user(username: str):
    if username not in ["alice", "bob"]:
        raise UserNotFoundException(username)
    return success_response(data={"username": username})

@app.post("/items/")
def create_item(item: dict):
    # 故意触发验证错误（无 Pydantic 模型）
    name = item["name"]  # 如果没有 name 字段会报错
    return success_response(data={"item": name})

---

六、练习任务（动手实践）

🧠 请先自己尝试完成，再查看下方答案！

任务1：添加“文章未找到”异常

• 定义 PostNotFoundException
• 注册异常处理器，返回 404 和友好消息
• 在 /posts/{post_id} 路由中使用（如果文章不存在）

任务2：限制 CORS 来源

• 修改 CORS 配置，只允许 http://localhost:3000（常见前端开发端口）

任务3（挑战）：统一错误响应格式

• 所有异常处理器返回 { "code": ..., "message": ..., "data": null }
• 确保 404、400、500 等都遵循此格式

---

七、练习任务参考答案

任务1 答案

# exceptions.py（新增）
class PostNotFoundException(Exception):
    def __init__(self, post_id: int):
        self.post_id = post_id

# main.py
from exceptions import PostNotFoundException

@app.exception_handler(PostNotFoundException)
async def post_not_found_handler(request: Request, exc: PostNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"code": 404, "message": f"文章 {exc.post_id} 不存在", "data": None}
    )

# 在 read_post 路由中
@app.get("/posts/{post_id}", response_model=Post)
def read_post(post_id: int, session: Session = Depends(get_session)):
    post = session.get(Post, post_id)
    if not post:
        raise PostNotFoundException(post_id)
    return post

---

任务2 答案

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # ← 修改这里
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

---

任务3 答案

确保所有异常处理器返回统一格式：

# 示例：HTTPException 也统一处理
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code,
            "message": exc.detail,
            "data": None
        }
    )

💡 注意：HTTPException 是 FastAPI 内部使用的异常，捕获它可统一 401、403 等。

---

八、小结

在本部分，你学会了：

• 使用 @app.exception_handler 自定义错误响应
• 通过 CORS 中间件 解决跨域问题
• 编写 自定义中间件 实现日志、性能监控
• 设计 统一响应格式，提升 API 专业性