FastAPI: 构建具有自动验证的 REST API

1. 概述

编写 REST API 时，大部分“防御”代码都在于检查输入：字段是否必填、数字是否为负、电子邮件是否有效。FastAPI 将这一职责交给 Pydantic。你只需声明数据类型和约束，框架就会自动验证请求并返回结构化的 422 错误。

本文从零开始构建一个订单管理 API。读完后，你将了解如何约束 body、path 和 query。你还将学会编写自定义验证器、过滤返回数据、修改错误格式，以及使用 pytest 测试所有内容。

2. 安装

你需要 Python 3.10 或更高版本。创建虚拟环境并安装 FastAPI 的 standard 依赖组：

python -m venv .venv
source .venv/bin/activate
pip install "fastapi[standard]" pytest

fastapi[standard] 包已包含本文所需的一切：

• uvicorn 用于运行服务器。
• fastapi-cli 提供 fastapi dev 命令。
• httpx 用于 TestClient。
• email-validator 用于 EmailStr 类型。

3. 第一个使用 Pydantic model 的 endpoint

创建文件 main.py：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Order API")


class OrderCreate(BaseModel):
    customer_email: str
    product_sku: str
    quantity: int


@app.post("/orders", status_code=201)
def create_order(order: OrderCreate) -> OrderCreate:
    return order

在开发模式下运行服务器，支持自动重新加载：

fastapi dev main.py

这里没有任何 if 语句。参数 order: OrderCreate 告诉 FastAPI 三件事：

• 从请求 body 中读取 JSON。
• 将 JSON 解析为 OrderCreate。
• 如果数据不匹配类型，返回 422。

尝试发送一个不是数字的 quantity：

curl -X POST localhost:8000/orders \
  -H "Content-Type: application/json" \
  -d '{"customer_email":"an@example.com","product_sku":"TSHIRT-01","quantity":"abc"}'

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

loc 字段指示错误的确切位置：位于 body 中的 quantity 字段。此外，/docs 处的 OpenAPI 文档是从这个 model 生成的。代码和文档因此永远不会偏离。

4. 使用 Field 进行数据约束

仅检查类型是不够的。例如 quantity = -5 仍然是有效的 int。使用 Field 添加约束，结合 Annotated：

from decimal import Decimal
from typing import Annotated

from pydantic import BaseModel, EmailStr, Field

Sku = Annotated[str, Field(min_length=3, max_length=20, pattern=r"^[A-Za-z0-9-]+$")]


class OrderCreate(BaseModel):
    customer_email: EmailStr
    product_sku: Sku
    quantity: Annotated[int, Field(gt=0, le=100)]
    unit_price: Annotated[Decimal, Field(gt=0, max_digits=10, decimal_places=2)]
    discount: Annotated[Decimal, Field(ge=0, decimal_places=2)] = Decimal("0")
    note: Annotated[str | None, Field(max_length=200)] = None

每一行的作用如下：

• EmailStr 检查实际的电子邮件格式，而不仅仅是检查是否含有 @ 字符。
• Sku 是一个可重用的类型。它要求长度为 3-20 个字符，只包含字母、数字和连字符。
• quantity 必须大于 0 且不超过 100。
• unit_price 使用 Decimal 而不是 float 以避免货币计算中的舍入错误。最多 10 个数字，小数点后 2 位。
• discount 和 note 有默认值，所以是可选字段。没有默认值的任何字段都是必需的。

发送 quantity: 0 会收到：

{"type":"greater_than","loc":["body","quantity"],"msg":"Input should be greater than 0","input":0,"ctx":{"gt":0}}

注意： Pydantic 默认以 lax 模式运行。字符串 "5" 会被强制转换为数字 5 而不会报错。如果想要严格拒绝，请在 model 中添加 model_config = ConfigDict(strict=True)，或为每个字段使用 Field(strict=True)。

5. 验证 path 和 query parameter

不仅 body，path 和 query 也通过相同的机制进行验证：

from fastapi import HTTPException, Path, Query

orders_db: dict[int, dict] = {}


@app.get("/orders/{order_id}")
def get_order(order_id: Annotated[int, Path(gt=0)]):
    if order_id not in orders_db:
        raise HTTPException(status_code=404, detail="Order not found")
    return orders_db[order_id]


@app.get("/orders")
def list_orders(
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=50)] = 10,
    sku: Annotated[str | None, Query(min_length=3)] = None,
):
    items = [o for o in orders_db.values() if sku is None or o["product_sku"] == sku]
    start = (page - 1) * size
    return items[start : start + size]

不同请求的结果：

• GET /orders/0 返回 422，因为 order_id 必须大于 0。
• GET /orders?size=500 返回 422，因为 size 最大为 50。
• GET /orders?page=2&size=5 正常运行。两个值 page 和 size 在进入函数前已被强制转换为 int。

区分 422 和 404。422 表示请求格式错误。404 是业务逻辑问题，所以你需要自己 raise HTTPException。

6. 自定义验证器

某些规则无法用 Field 表达。此时使用 field_validator 验证单个字段，或 model_validator 验证涉及多个字段的规则。

from pydantic import field_validator, model_validator

FORBIDDEN_DOMAINS = {"mailinator.com", "tempmail.com"}


class OrderCreate(BaseModel):
    # ... 第 4 部分中的字段 ...

    @field_validator("product_sku")
    @classmethod
    def normalize_sku(cls, value: str) -> str:
        return value.upper()

    @field_validator("customer_email")
    @classmethod
    def reject_disposable_email(cls, value: str) -> str:
        domain = value.split("@")[-1].lower()
        if domain in FORBIDDEN_DOMAINS:
            raise ValueError("Disposable email is not allowed")
        return value

    @model_validator(mode="after")
    def check_discount(self) -> "OrderCreate":
        total = self.unit_price * self.quantity
        if self.discount > total:
            raise ValueError("Discount cannot exceed order total")
        return self

每个验证器的说明：

• normalize_sku 在 Field 验证 pattern 后运行。它将 SKU 标准化为大写，所以 "tshirt-01" 被保存为 "TSHIRT-01"。验证器可以转换值，只要 return 新值。
• reject_disposable_email 抛出 ValueError。Pydantic 将此错误转换为验证错误，消息以 "Value error, " 为前缀。
• check_discount 以 mode="after" 运行，即在所有字段都有效且具有正确类型时。这样你可以安全地乘以 Decimal。记住 return self。

注意： 不要在验证器中调用数据库或外部 API。例如，检查“SKU 是否在库存中”是业务逻辑。应该将其放在服务或依赖中。验证器应该是纯函数，快速且易于测试。

7. response_model：控制返回的数据

你通常存储的信息比返回给 client 的要多，例如内部备注。分离输入和输出 model：

class OrderRead(BaseModel):
    id: int
    customer_email: EmailStr
    product_sku: str
    quantity: int
    total: Decimal


@app.post("/orders", status_code=201, response_model=OrderRead)
def create_order(order: OrderCreate):
    order_id = len(orders_db) + 1
    record = {
        "id": order_id,
        **order.model_dump(),
        "total": order.unit_price * order.quantity - order.discount,
        "internal_flag": "manual-review",
    }
    orders_db[order_id] = record
    return record

函数返回包含 note 和 internal_flag 的 dict。但是 response_model=OrderRead 过滤响应以仅包含已声明的字段：

{"id":1,"customer_email":"an@example.com","product_sku":"TSHIRT-01","quantity":2,"total":"39.98"}

有两点需要注意：

• Decimal 被序列化为字符串 "39.98" 以保持精度。Client 需要知道这一点。
• 响应也会被验证。如果你忘记了 total 字段，FastAPI 会抛出 ResponseValidationError 并返回 500，而不是 422。这是服务器的错误，而不是 client 的错误。

8. 自定义 422 错误格式

默认格式相当详细。许多团队希望 frontend 有一个简洁统一的结构。重写 RequestValidationError 的 handler：

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


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = [
        {
            "field": ".".join(str(part) for part in err["loc"][1:]) or "body",
            "message": err["msg"],
        }
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=422,
        content={"code": "VALIDATION_FAILED", "errors": errors},
    )

handler 处理每个错误的方式：

• err["loc"][1:] 移除第一个元素（body、query、path）并用点连接剩余部分。
• 来自 model_validator 的错误不会绑定到任何字段。此时 loc 只是 ("body",)，所以 handler 分配 "body"。

发送一个包含 quantity: 0 和一次性电子邮件的订单会收到：

{"code":"VALIDATION_FAILED","errors":[
  {"field":"customer_email","message":"Value error, Disposable email is not allowed"},
  {"field":"quantity","message":"Input should be greater than 0"}]}

Pydantic 在一次请求中收集所有字段的错误，而不是在第一个错误处停止。Client 因此可以一次性修复所有问题。

9. 使用 pytest 进行测试

TestClient 直接调用 app，无需运行服务器。在 main.py 旁边创建 test_main.py：

import pytest
from fastapi.testclient import TestClient

from main import app, orders_db

client = TestClient(app)

VALID_ORDER = {
    "customer_email": "an@example.com",
    "product_sku": "tshirt-01",
    "quantity": 2,
    "unit_price": "19.99",
}


@pytest.fixture(autouse=True)
def clear_db():
    orders_db.clear()


def test_create_order_returns_normalized_sku_and_total():
    response = client.post("/orders", json=VALID_ORDER)

    assert response.status_code == 201
    body = response.json()
    assert body["product_sku"] == "TSHIRT-01"
    assert body["total"] == "39.98"
    assert "internal_flag" not in body


def test_invalid_quantity_and_email_are_reported_together():
    payload = {**VALID_ORDER, "quantity": 0, "customer_email": "x@mailinator.com"}

    response = client.post("/orders", json=payload)

    assert response.status_code == 422
    fields = {e["field"] for e in response.json()["errors"]}
    assert fields == {"quantity", "customer_email"}


def test_discount_greater_than_total_is_rejected():
    payload = {**VALID_ORDER, "discount": "100.00"}

    response = client.post("/orders", json=payload)

    assert response.status_code == 422
    assert response.json()["errors"][0] == {
        "field": "body",
        "message": "Value error, Discount cannot exceed order total",
    }


def test_page_size_over_limit_is_rejected():
    assert client.get("/orders?size=500").status_code == 422

运行测试：

pytest -q

....                                                    [100%]
4 passed in 0.31s

fixture clear_db 在每个测试之前重置 orders_db，帮助测试之间相互独立。此外，你可以单独测试 model 而不需要 HTTP，使用 OrderCreate.model_validate(...) 结合 pytest.raises(ValidationError)。当 model 有许多复杂规则时，这种方法很合适。

10. 常见错误

• 使用 float 表示货币。 0.1 + 0.2 不等于 0.3。使用带 max_digits 和 decimal_places 的 Decimal。
• 忘记 @classmethod 或 decorator 的顺序。 @field_validator 必须在 @classmethod 上面。否则 Pydantic v2 在导入时报错。
• 在验证器中忘记 return。 field_validator 不返回会将 None 分配给字段。model_validator(mode="after") 必须返回 self。
• 使用 Pydantic v1 语法。 @validator、class Config、.dict() 已弃用。使用 @field_validator、model_config、.model_dump()。
• 混淆 422 和 500。 输入错误返回 422。响应不符合 response_model 返回 500。看到 500 时，检查你返回的数据。
• 对输入和输出使用同一个 model。 这样容易泄露内部字段，并强制 client 发送 id 等字段。分离 OrderCreate 和 OrderRead。

注意： API 层的验证不能替代数据库约束。UNIQUE 或 NOT NULL 等约束仍然是必要的。请求可能通过多条路径进入系统，如后台作业或迁移脚本。

11. 结论

FastAPI 与 Pydantic 的结合将类型提示转变为数据契约，有三个主要优势：

• 错误的请求在入口处被拦截。
• 错误返回时完整且有结构。
• OpenAPI 文档总是与代码匹配。

在本文中，你使用了 Field 进行简单约束，Path 和 Query 进行 URL 参数约束，field_validator 和 model_validator 进行自定义规则。你还使用了 response_model 过滤输出，修改了 422 错误格式，并使用 TestClient 测试了全部内容。

记住的原则：让验证器检查数据的形状，而将业务逻辑放在服务层。