第六章中间件与异常处理

本章目标：掌握 FastAPI 的异常处理机制（HTTPException、自定义异常处理器、全局校验错误捕获），以及中间件的概念和常用实践（请求耗时记录、CORS 跨域配置）。
预计时长：40 分钟

6.1 异常处理

HTTPException 的使用

在前面的章节中我们已经用过 HTTPException，这里系统整理它的用法。

HTTPException 是 FastAPI 提供的标准异常类，抛出后会立即终止请求并返回指定的 HTTP 错误响应。

python

from fastapi import FastAPI, HTTPException

app = FastAPI(title="异常处理示例")

fake_users = {
    1: {"id": 1, "name": "张三", "status": "active"},
    2: {"id": 2, "name": "李四", "status": "banned"},
}


@app.get("/users/{user_id}")
def get_user(user_id: int):
    if user_id not in fake_users:
        raise HTTPException(status_code=404, detail="用户不存在")

    user = fake_users[user_id]
    if user["status"] == "banned":
        raise HTTPException(
            status_code=403,
            detail="该用户已被封禁",
            headers={"X-Error-Reason": "user-banned"},
        )
    return user

HTTPException 的参数：

参数	类型	说明
`status_code`	`int`	HTTP 状态码（404、403、500 等）
`detail`	`str / dict / list`	错误详情，会出现在响应 JSON 的 `detail` 字段
`headers`	`dict`	可选，附加到错误响应的自定义响应头

detail 支持复杂结构：

python

raise HTTPException(
    status_code=400,
    detail={
        "code": "INVALID_ORDER",
        "message": "订单校验失败",
        "errors": [
            {"field": "quantity", "message": "库存不足"},
            {"field": "coupon", "message": "优惠券已过期"},
        ],
    },
)

响应：

json

{
    "detail": {
        "code": "INVALID_ORDER",
        "message": "订单校验失败",
        "errors": [
            {"field": "quantity", "message": "库存不足"},
            {"field": "coupon", "message": "优惠券已过期"}
        ]
    }
}

常用 HTTP 状态码速查：

状态码	含义	典型场景
400	Bad Request	业务逻辑校验失败
401	Unauthorized	未登录 / Token 无效
403	Forbidden	已登录但无权限
404	Not Found	资源不存在
409	Conflict	资源冲突（如用户名已存在）
422	Unprocessable Entity	请求参数校验失败（FastAPI 自动返回）
500	Internal Server Error	服务器内部错误

自定义异常处理器

有时候你想用自己的异常类来标识特定的错误类型，并统一处理它们。

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI(title="自定义异常处理器")


class BusinessError(Exception):
    """自定义业务异常"""

    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code


@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error_code": exc.code,
            "message": exc.message,
        },
    )


fake_products = {"P001": {"id": "P001", "name": "键盘", "stock": 5}}


@app.post("/orders")
def create_order(product_id: str, quantity: int):
    product = fake_products.get(product_id)
    if not product:
        raise BusinessError(
            code="PRODUCT_NOT_FOUND",
            message=f"商品 {product_id} 不存在",
            status_code=404,
        )
    if quantity > product["stock"]:
        raise BusinessError(
            code="INSUFFICIENT_STOCK",
            message=f"库存不足，当前库存 {product['stock']}，请求 {quantity}",
        )
    return {"message": "下单成功", "product": product["name"], "quantity": quantity}

执行流程：

路由函数中 raise BusinessError(...)
  │
  ▼
FastAPI 捕获异常，查找对应的 exception_handler
  │
  ▼
business_error_handler(request, exc) 执行
  │
  ▼
返回自定义格式的 JSONResponse

测试：

请求	预期响应
`POST /orders?product_id=P001&quantity=3`	`200` — 下单成功
`POST /orders?product_id=P999&quantity=1`	`404` — `{"success": false, "error_code": "PRODUCT_NOT_FOUND", ...}`
`POST /orders?product_id=P001&quantity=100`	`400` — `{"success": false, "error_code": "INSUFFICIENT_STOCK", ...}`

RequestValidationError 的全局捕获

FastAPI 在参数校验失败时自动返回 422 错误。默认的错误格式较冗长，你可能想统一格式：

默认的 422 响应：

json

{
    "detail": [
        {
            "type": "int_parsing",
            "loc": ["path", "user_id"],
            "msg": "Input should be a valid integer, unable to parse string as an integer",
            "input": "abc"
        }
    ]
}

自定义后的 422 响应：

json

{
    "success": false,
    "error_code": "VALIDATION_ERROR",
    "message": "请求参数校验失败",
    "errors": [
        {
            "field": "path → user_id",
            "message": "Input should be a valid integer, unable to parse string as an integer"
        }
    ]
}

实现代码：

python

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="全局校验错误捕获")


@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field_path = " → ".join(str(loc) for loc in error["loc"])
        errors.append({
            "field": field_path,
            "message": error["msg"],
        })
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error_code": "VALIDATION_ERROR",
            "message": "请求参数校验失败",
            "errors": errors,
        },
    )


class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    age: int = Field(ge=1, le=150)
    email: str


@app.post("/users")
def create_user(user: UserCreate):
    return {"message": "创建成功", "user": user}


@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": "测试用户"}

bash

fastapi dev chapter06.py

测试：

请求	预期
`GET /users/abc`	422 — 统一格式的校验错误（user_id 不是整数）
`POST /users` body: `{"username": "ab", "age": 200, "email": "test"}`	422 — 多个字段的校验错误

整合：统一异常处理模板

在实际项目中，建议把所有异常处理器集中管理：

python

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

app = FastAPI(title="统一异常处理")


class BusinessError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code


@app.exception_handler(BusinessError)
async def handle_business_error(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"success": False, "error_code": exc.code, "message": exc.message},
    )


@app.exception_handler(RequestValidationError)
async def handle_validation_error(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field_path = " → ".join(str(loc) for loc in error["loc"])
        errors.append({"field": field_path, "message": error["msg"]})
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error_code": "VALIDATION_ERROR",
            "message": "请求参数校验失败",
            "errors": errors,
        },
    )


@app.exception_handler(Exception)
async def handle_unexpected_error(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error_code": "INTERNAL_ERROR",
            "message": "服务器内部错误，请稍后重试",
        },
    )

这三个处理器的覆盖范围：

异常类型                    处理器                    场景
HTTPException             FastAPI 内置处理          raise HTTPException(...)
BusinessError             handle_business_error     raise BusinessError(...)
RequestValidationError    handle_validation_error   参数校验失败（自动触发）
Exception                 handle_unexpected_error   未预料的异常（兜底）

6.2 中间件

什么是中间件

中间件是请求和响应的拦截器：每个请求在到达路由函数之前、每个响应在返回客户端之前，都会经过中间件。

客户端请求
  │
  ▼
┌─────────────────────────────────┐
│         中间件 A（最外层）        │
│  ┌───────────────────────────┐  │
│  │      中间件 B（内层）       │  │
│  │  ┌─────────────────────┐  │  │
│  │  │    路由函数处理请求    │  │  │
│  │  └─────────────────────┘  │  │
│  └───────────────────────────┘  │
└─────────────────────────────────┘
  │
  ▼
客户端响应

中间件按「洋葱模型」执行：

请求从外到内依次经过每层中间件
到达路由函数，处理业务逻辑
响应从内到外依次经过每层中间件

自定义中间件：记录请求耗时

最常见的中间件需求——记录每个请求的处理时间：

python

import time
from fastapi import FastAPI, Request

app = FastAPI(title="中间件示例")


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()

    response = await call_next(request)

    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}s"
    print(f"{request.method} {request.url.path} - {process_time:.4f}s")

    return response


@app.get("/fast")
def fast_endpoint():
    return {"message": "这个接口很快"}


@app.get("/slow")
def slow_endpoint():
    time.sleep(1)
    return {"message": "这个接口模拟了 1 秒延迟"}

bash

fastapi dev chapter06_middleware.py

访问接口后，观察：

控制台输出：GET /fast - 0.0003s、GET /slow - 1.0012s
响应头中包含：X-Process-Time: 0.0003s

中间件函数的结构：

python

@app.middleware("http")
async def my_middleware(request: Request, call_next):
    # ① 请求到达路由之前的逻辑
    # ...

    response = await call_next(request)  # ② 调用后续中间件/路由

    # ③ 响应返回客户端之前的逻辑
    # ...

    return response

阶段	位置	典型用途
①请求前	`call_next` 之前	日志记录、认证检查、请求改写
②处理	`call_next(request)`	将请求传递给下一层
③响应后	`call_next` 之后	添加响应头、记录耗时、响应改写

更实用的中间件：请求日志

python

import time
from fastapi import FastAPI, Request

app = FastAPI(title="请求日志中间件")


@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.perf_counter()

    client_ip = request.client.host if request.client else "unknown"
    print(f"→ {request.method} {request.url.path} from {client_ip}")

    response = await call_next(request)

    duration = time.perf_counter() - start_time
    print(f"← {request.method} {request.url.path} → {response.status_code} ({duration:.4f}s)")

    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    response.headers["X-Request-ID"] = str(id(request))

    return response


@app.get("/hello")
def hello():
    return {"message": "Hello!"}

控制台输出效果：

→ GET /hello from 127.0.0.1
← GET /hello → 200 (0.0005s)

CORS 中间件配置（前后端分离必备）

当前端（如 http://localhost:3000）和后端（如 http://localhost:8000）不在同一个域时，浏览器会阻止跨域请求。这就是 CORS（Cross-Origin Resource Sharing）问题。

前端 http://localhost:3000
  │
  │ fetch("http://localhost:8000/api/users")
  │
  ▼
浏览器: "跨域了！先问后端允不允许..."
  │
  │ OPTIONS http://localhost:8000/api/users  ← 预检请求
  │
  ▼
后端没配 CORS → 浏览器拒绝请求 ❌
后端配了 CORS → 浏览器允许请求 ✅

FastAPI 内置了 CORS 中间件，配置简单：

python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="CORS 示例")

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",
        "http://localhost:5173",
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/api/users")
def list_users():
    return [{"id": 1, "name": "张三"}, {"id": 2, "name": "李四"}]

CORS 中间件参数说明：

参数	说明	推荐值
`allow_origins`	允许的来源域名列表	生产环境明确列出；开发可用 `["*"]`
`allow_credentials`	是否允许发送 Cookie	`True`（如果用 Cookie 认证）
`allow_methods`	允许的 HTTP 方法	`["*"]` 或 `["GET", "POST", "PUT", "DELETE"]`
`allow_headers`	允许的请求头	`["*"]` 或明确列出
`max_age`	预检请求缓存时间（秒）	`600`（10 分钟）

安全提示：生产环境不要用 allow_origins=["*"]，应该明确列出允许的前端域名。

完整示例：组合中间件 + 异常处理

python

import time
from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="中间件 + 异常处理综合示例")

# ---------- CORS 中间件 ----------
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ---------- 请求耗时中间件 ----------
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    print(f"{request.method} {request.url.path} → {response.status_code} ({duration:.4f}s)")
    return response


# ---------- 自定义异常 ----------
class BusinessError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code


@app.exception_handler(BusinessError)
async def handle_business_error(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"success": False, "error_code": exc.code, "message": exc.message},
    )


@app.exception_handler(RequestValidationError)
async def handle_validation_error(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field_path = " → ".join(str(loc) for loc in error["loc"])
        errors.append({"field": field_path, "message": error["msg"]})
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error_code": "VALIDATION_ERROR",
            "message": "请求参数校验失败",
            "errors": errors,
        },
    )


# ---------- 数据模型 ----------
class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    quantity: int = Field(ge=0, default=0)


items_db: dict[int, dict] = {}
next_id = 1


# ---------- 接口 ----------
@app.post("/items", status_code=201)
def create_item(item: ItemCreate):
    global next_id
    for existing in items_db.values():
        if existing["name"] == item.name:
            raise BusinessError(
                code="DUPLICATE_ITEM",
                message=f"商品 '{item.name}' 已存在",
                status_code=409,
            )
    new_item = {"id": next_id, **item.model_dump()}
    items_db[next_id] = new_item
    next_id += 1
    return new_item


@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    return items_db[item_id]


@app.get("/items")
def list_items():
    return list(items_db.values())

bash

fastapi dev chapter06_full.py

测试清单：

测试	操作	预期
正常创建	`POST /items` → `{"name": "键盘", "price": 299, "quantity": 5}`	201 + 响应头含 `X-Process-Time`
重复创建	再次 POST 相同 name	409 — `DUPLICATE_ITEM`
校验失败	`POST /items` → `{"name": "", "price": -1}`	422 — 统一格式的错误
不存在	`GET /items/999`	404 — 商品不存在
控制台	观察终端	每个请求都打印了路径和耗时

6.3 动手练习

练习 1：添加请求 ID 中间件

实现一个中间件，为每个请求生成唯一 ID，并在响应头中返回：

python

import uuid
import time
from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    start = time.perf_counter()

    response = await call_next(request)

    duration = time.perf_counter() - start
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    print(f"[{request_id[:8]}] {request.method} {request.url.path} → {response.status_code} ({duration:.4f}s)")

    return response


@app.get("/hello")
def hello():
    return {"message": "Hello!"}

启动后多次访问 /hello，观察：

每次请求的 X-Request-ID 不同
控制台输出带有请求 ID 前缀，方便排查问题

练习 2：IP 黑名单中间件

实现一个中间件，拦截黑名单中的 IP 地址：

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

BLOCKED_IPS = {"192.168.1.100", "10.0.0.50"}


@app.middleware("http")
async def block_blacklisted_ips(request: Request, call_next):
    client_ip = request.client.host if request.client else "unknown"
    if client_ip in BLOCKED_IPS:
        return JSONResponse(
            status_code=403,
            content={"detail": "你的 IP 已被封禁"},
        )
    return await call_next(request)


@app.get("/hello")
def hello():
    return {"message": "Hello!"}

提示：中间件中可以直接返回 Response 而不调用 call_next，这样请求就不会到达路由函数。

本章小结

概念	要点
`HTTPException`	FastAPI 内置异常，指定 `status_code` 和 `detail` 即可返回错误
`detail` 结构	支持字符串、字典、列表，可构造复杂错误信息
`@app.exception_handler()`	注册自定义异常处理器，统一错误响应格式
`RequestValidationError`	参数校验失败时自动触发，可全局捕获并自定义格式
兜底异常处理	捕获 `Exception` 防止未预料错误暴露堆栈信息
中间件	请求/响应的拦截器，按洋葱模型执行
`@app.middleware("http")`	自定义中间件的装饰器写法
`call_next(request)`	调用下一层中间件或路由函数
CORS 中间件	`CORSMiddleware`，前后端分离项目必配
`allow_origins`	生产环境应明确列出允许的前端域名，不要用 `["*"]`

下一章预告：我们将学习 FastAPI 的异步编程——理解 async/await 在 FastAPI 中的正确用法，以及什么场景用同步、什么场景用异步能获得最佳性能。

第六章 中间件与异常处理 ​

6.1 异常处理 ​

HTTPException 的使用 ​

自定义异常处理器 ​

RequestValidationError 的全局捕获 ​

整合：统一异常处理模板 ​

6.2 中间件 ​

什么是中间件 ​

自定义中间件：记录请求耗时 ​

更实用的中间件：请求日志 ​

CORS 中间件配置（前后端分离必备） ​

完整示例：组合中间件 + 异常处理 ​

6.3 动手练习 ​

练习 1：添加请求 ID 中间件 ​

练习 2：IP 黑名单中间件 ​

本章小结 ​

第六章中间件与异常处理

6.1 异常处理

HTTPException 的使用

自定义异常处理器

RequestValidationError 的全局捕获

整合：统一异常处理模板

6.2 中间件

什么是中间件

自定义中间件：记录请求耗时

更实用的中间件：请求日志

CORS 中间件配置（前后端分离必备）

完整示例：组合中间件 + 异常处理

6.3 动手练习

练习 1：添加请求 ID 中间件

练习 2：IP 黑名单中间件

本章小结