数据验证技能 data-validation

数据验证技能

当实现API有效载荷、表单输入或数据库写入的数据验证时使用。触发器包括：Pydantic模型、Zod模式、输入清理、类型验证、字段约束或请求/响应模式。不包括：业务逻辑（使用领域服务）或认证/授权。

快速参考

层	工具	目的
后端	Pydantic v2	请求/响应验证
前端	Zod	表单验证、模式推断
清理	bleach, strip-html	输入清理
类型	mypy, TypeScript	编译时安全

项目结构

backend/
├── app/
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── student.py      # StudentCreate, StudentUpdate, StudentOut
│   │   ├── fee.py          # FeeCreate, FeeOut
│   │   └── common.py       # PaginationParams, ErrorDetail
│   └── models/             # SQLModel （与模式分开）
frontend/
├── lib/
│   ├── schemas/
│   │   ├── student.schema.ts
│   │   └── index.ts        # 导出所有模式
│   └── validation.ts       # 共享验证工具
└── types/
    └── shared.ts           # 从Zod推断出的类型

后端：Pydantic模型

分离输入/输出模型

# backend/app/schemas/student.py
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, EmailStr, Field, ConfigDict

# === CREATE Models (Input) ===

class StudentCreate(BaseModel):
    """创建新学生的负载。"""

    first_name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        examples=["John"],
        description="学生的名",
    )
    last_name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        examples=["Doe"],
    )
    email: EmailStr = Field(..., examples=["john.doe@school.edu"])
    phone: Optional[str] = Field(
        None,
        pattern=r"^\+?[1-9]\d{9,14}$",
        examples=["+1234567890"],
    )
    date_of_birth: datetime = Field(..., description="学生的生日")
    grade_level: int = Field(..., ge=1, le=12, examples=[9])
    enrollment_date: datetime = Field(default_factory=datetime.utcnow)

    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "first_name": "John",
                    "last_name": "Doe",
                    "email": "john.doe@school.edu",
                    "date_of_birth": "2008-05-15T00:00:00Z",
                    "grade_level": 9,
                }
            ]
        }
    )

class StudentUpdate(BaseModel):
    """更新学生信息的负载。所有字段可选。"""

    first_name: Optional[str] = Field(None, min_length=1, max_length=100)
    last_name: Optional[str] = Field(None, min_length=1, max_length=100)
    email: Optional[EmailStr] = None
    phone: Optional[str] = Field(None, pattern=r"^\+?[1-9]\d{9,14}$")
    grade_level: Optional[int] = Field(None, ge=1, le=12)
    is_active: Optional[bool] = None

# === OUTPUT Models (Response) ===

class StudentOut(BaseModel):
    """学生数据的响应模型。"""

    id: int
    first_name: str
    last_name: str
    email: str
    phone: Optional[str] = None
    date_of_birth: datetime
    grade_level: int
    enrollment_date: datetime
    is_active: bool
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)

class StudentWithFees(StudentOut):
    """带有嵌套费用信息的学生。"""

    fees: list["FeeOut"] = []

自定义验证器

# backend/app/schemas/student.py （继续）
from pydantic import field_validator, model_validator
from datetime import date

class StudentCreate(BaseModel):
    # ... 上面的字段

    @field_validator("date_of_birth", mode="before")
    @classmethod
    def parse_date_of_birth(cls, v):
        if isinstance(v, str):
            return datetime.fromisoformat(v.replace("Z", "+00:00"))
        return v

    @model_validator(mode="after")
    def validate_age_range(self):
        """确保学生是合理的学龄。"""
        dob = self.date_of_birth
        if isinstance(dob, datetime):
            dob = dob.date()

        today = date.today()
        age = today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day))

        if age < 5:
            raise ValueError("学生必须至少5岁")
        if age > 25:
            raise ValueError("年龄对于学校注册来说似乎不合理")
        return self

通用模式

# backend/app/schemas/common.py
from typing import Generic, TypeVar, Optional, List
from pydantic import BaseModel, Field


T = TypeVar("T")


class PaginationParams(BaseModel):
    """标准分页参数。"""

    skip: int = Field(0, ge=0, description="要跳过的记录数")
    limit: int = Field(100, ge=1, le=1000, description="返回的最大记录数")


class PaginatedResponse(BaseModel, Generic[T]):
    """标准分页响应包装器。"""

    data: List[T]
    total: int = Field(..., description="记录的总数")
    skip: int
    limit: int
    has_more: bool = Field(..., description="是否还有更多记录")


class ErrorDetail(BaseModel):
    """详细的验证错误。"""

    field: str = Field(..., examples=["email"])
    message: str = Field(..., examples=["无效的电子邮件格式"])
    code: str = Field(..., examples=["value_error"])


class ValidationErrorResponse(BaseModel):
    """标准验证错误响应。"""

    error: dict = Field(..., description="错误详情")
    details: Optional[List[ErrorDetail]] = None

# === 搜索和过滤 ===

class SearchParams(BaseModel):
    """标准搜索参数。"""

    query: Optional[str] = Field(None, min_length=1, max_length=200)
    sort_by: str = Field("created_at")
    sort_order: str = Field("desc", pattern="^(asc|desc)$")

前端：Zod模式

基本模式

// frontend/lib/schemas/student.schema.ts
import { z } from "zod";

// === 常量（与后端共享） ===
const NAME_MIN_LENGTH = 1;
const NAME_MAX_LENGTH = 100;
const PHONE_REGEX = /^\+?[1-9]\d{9,14}$/;
const GRADE_MIN = 1;
const GRADE_MAX = 12;

// === CREATE SCHEMA ===
export const studentCreateSchema = z.object({
  first_name: z
    .string()
    .min(NAME_MIN_LENGTH, { message: "名字是必填项" })
    .max(NAME_MAX_LENGTH, { message: "允许的最大字符数为100" })
    .trim(),

  last_name: z
    .string()
    .min(NAME_MIN_LENGTH, { message: "姓氏是必填项" })
    .max(NAME_MAX_LENGTH, { message: "允许的最大字符数为100" })
    .trim(),

  email: z.string().email({ message: "无效的电子邮件地址" }),

  phone: z
    .string()
    .regex(PHONE_REGEX, { message: "无效的电话号码格式" })
    .optional()
    .or(z.literal("")),

  date_of_birth: z.string().datetime({ message: "无效的日期格式" }),

  grade_level: z
    .number()
    .int()
    .min(GRADE_MIN, { message: `年级至少为${GRADE_MIN}` })
    .max(GRADE_MAX, { message: `年级不能超过${GRADE_MAX}` }),
});

// === UPDATE SCHEMA （部分） ===
export const studentUpdateSchema = studentCreateSchema.partial();

// === OUTPUT/RESPONSE SCHEMA ===
export const studentOutSchema = z.object({
  id: z.number(),
  first_name: z.string(),
  last_name: z.string(),
  email: z.string().email(),
  phone: z.string().nullable(),
  date_of_birth: z.string().datetime(),
  grade_level: z.number().int().min(GRADE_MIN).max(GRADE_MAX),
  enrollment_date: z.string().datetime(),
  is_active: z.boolean(),
  created_at: z.string().datetime(),
  updated_at: z.string().datetime(),
});

// === 推断出的类型 ===
export type StudentCreateInput = z.infer<typeof studentCreateSchema>;
export type StudentUpdateInput = z.infer<typeof studentUpdateSchema>;
export type StudentOutput = z.infer<typeof studentOutSchema>;

高级Zod模式

// frontend/lib/schemas/student.schema.ts （继续）
import { date, string } from "zod";

// === 跨字段验证 ===
export const studentCreateSchema = z
  .object({
    first_name: z.string().min(1).max(100),
    last_name: z.string().min(1).max(100),
    email: z.string().email(),
    phone: z.string().regex(/^\+?[1-9]\d{9,14}$/).optional(),
    date_of_birth: z.string().datetime(),
    grade_level: z.number().int().min(1).max(12),
  })
  .refine(
    (data) => {
      // 跨字段验证：出生日期必须在过去
      const dob = new Date(data.date_of_birth);
      return dob < new Date();
    },
    {
      message: "出生日期必须在过去",
      path: ["date_of_birth"],
    }
  )
  .refine(
    (data) => {
      // 确保年龄适合上学
      const dob = new Date(data.date_of_birth);
      const age =
        new Date().getFullYear() - dob.getFullYear();
      return age >= 5 && age <= 25;
    },
    {
      message: "学生年龄必须在5到25岁之间",
      path: ["date_of_birth"],
    }
  );

// === 自定义转换器 ===
export const phoneSchema = z
  .string()
  .regex(/^\+?[1-9]\d{9,14}$/, { message: "无效的电话号码" })
  .transform((phone) => phone.replace(/\s+/g, "")); // 规范化

// === 枚举模式 ===
export const feeStatusSchema = z.enum(["pending", "paid", "overdue", "waived"]);

// === 嵌套模式 ===
export const addressSchema = z.object({
  street: z.string().min(1).max(200),
  city: z.string().min(1).max(100),
  state: z.string().min(2).max(50),
  zip_code: z.string().regex(/^\d{5}(-\d{4})?$/),
  country: z.string().min(2).max(100).default("USA"),
});

export const studentWithAddressSchema = studentOutSchema.extend({
  address: addressSchema.nullable(),
});

// === 区分联合 ===
export const enrollmentStatusSchema = z.discriminatedUnion("status", [
  z.object({
    status: z.literal("active"),
    last_attendance_date: z.string().datetime(),
  }),
  z.object({
    status: z.literal("inactive"),
    reason: z.string().min(1),
    inactive_since: z.string().datetime(),
  }),
  z.object({
    status: z.literal("pending"),
    expected_start_date: z.string().datetime(),
  }),
]);

表单集成

// frontend/components/StudentForm.tsx
"use client";

import { useForm } from "react-hook-form";
import { zodResolver } from "@hookform/resolvers/zod";
import { studentCreateSchema, type StudentCreateInput } from "@/lib/schemas/student.schema";

export function StudentForm() {
  const {
    register,
    handleSubmit,
    formState: { errors, isSubmitting },
    setError,
  } = useForm<StudentCreateInput>({
    resolver: zodResolver(studentCreateSchema),
    mode: "onBlur",
  });

  const onSubmit = async (data: StudentCreateInput) => {
    try {
      const response = await fetch("/api/v1/students", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(data),
      });

      if (!response.ok) {
        const errorData = await response.json();

        // 将后端错误映射到表单字段
        if (errorData.details) {
          for (const detail of errorData.details) {
            setError(detail.field as keyof StudentCreateInput, {
              message: detail.message,
            });
          }
        }
        return;
      }

      // 成功处理
      router.push("/students");
    } catch {
      // 网络或意外错误
      setError("root", {
        message: "发生了一个意外的错误。请再试一次。",
      });
    }
  };

  return (
    <form onSubmit={handleSubmit(onSubmit)} className="space-y-4">
      {/* 名字 */}
      <div>
        <label htmlFor="first_name">名字</label>
        <input
          id="first_name"
          {...register("first_name")}
          className={errors.first_name ? "border-red-500" : ""}
        />
        {errors.first_name && (
          <p className="text-red-500">{errors.first_name.message}</p>
        )}
      </div>

      {/* 姓氏 */}
      <div>
        <label htmlFor="last_name">姓氏</label>
        <input
          id="last_name"
          {...register("last_name")}
          className={errors.last_name ? "border-red-500" : ""}
        />
        {errors.last_name && (
          <p className="text-red-500">{errors.last_name.message}</p>
        )}
      </div>

      {/* 电子邮件 */}
      <div>
        <label htmlFor="email">电子邮件</label>
        <input
          id="email"
          type="email"
          {...register("email")}
          className={errors.email ? "border-red-500" : ""}
        />
        {errors.email && (
          <p className="text-red-500">{errors.email.message}</p>
        )}
      </div>

      {/* 年级水平 */}
      <div>
        <label htmlFor="grade_level">年级水平</label>
        <select
          id="grade_level"
          {...register("grade_level", { valueAsNumber: true })}
          className={errors.grade_level ? "border-red-500" : ""}
        >
          {Array.from({ length: 12 }, (_, i) => i + 1).map((grade) => (
            <option key={grade} value={grade}>
              年级 {grade}
            </option>
          ))}
        </select>
        {errors.grade_level && (
          <p className="text-red-500">{errors.grade_level.message}</p>
        )}
      </div>

      {/* 出生日期 */}
      <div>
        <label htmlFor="date_of_birth">出生日期</label>
        <input
          id="date_of_birth"
          type="date"
          {...register("date_of_birth")}
          className={errors.date_of_birth ? "border-red-500" : ""}
        />
        {errors.date_of_birth && (
          <p className="text-red-500">{errors.date_of_birth.message}</p>
        )}
      </div>

      {errors.root && (
        <p className="text-red-500">{errors.root.message}</p>
      )}

      <button type="submit" disabled={isSubmitting}>
        {isSubmitting ? "创建中..." : "创建学生"}
      </button>
    </form>
  );
}

输入清理

后端清理

# backend/app/core/sanitize.py
import re
from typing import Any


def strip_html(text: str) -> str:
    """从文本中移除HTML标签。"""
    if not text:
        return text
    # 简单的剥离 - 使用bleach进行更健壮的清理
    return re.sub(r"<[^>]*>", "", text)


def sanitize_string(
    value: Any,
    max_length: int = None,
    allow_html: bool = False,
) -> str:
    """清理字符串值。"""
    if value is None:
        return ""

    s = str(value).strip()

    if not allow_html:
        s = strip_html(s)

    if max_length:
        s = s[:max_length]

    return s


def sanitize_for_database(value: Any) -> Any:
    """通用数据库输入清理。"""
    if isinstance(value, str):
        return sanitize_string(value, max_length=10000)
    if isinstance(value, list):
        return [sanitize_for_database(v) for v in value]
    if isinstance(value, dict):
        return {k: sanitize_for_database(v) for k, v in value.items()}
    return value

带清理的验证器

# backend/app/schemas/student.py
from pydantic import field_validator


class StudentCreate(BaseModel):
    first_name: str
    last_name: str
    email: EmailStr

    @field_validator("first_name", "last_name", mode="before")
    @classmethod
    def sanitize_names(cls, v):
        if isinstance(v, str):
            return sanitize_string(v, max_length=100)
        return v

模式共享策略

字段一致性

# backend/app/schemas/student.py
class StudentCreate(BaseModel):
    first_name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        json_schema_extra={
            "frontend_type": "text",
            "validation": {
                "min": 1,
                "max": 100,
            },
        },
    )

// frontend/lib/schemas/shared/constants.ts
// 与后端约束匹配
export const STUDENT_NAME_MAX_LENGTH = 100;
export const STUDENT_NAME_MIN_LENGTH = 1;
export const PHONE_REGEX = /^\+?[1-9]\d{9,14}$/;
export const GRADE_MIN = 1;
export const GRADE_MAX = 12;

合同文档

# 学生模式合同

## StudentCreate

| 字段 | 类型 | 必需 | 最小 | 最大 | 模式 | 描述 |
|-------|------|----------|-----|-----|---------|-------------|
| first_name | string | 是 | 1 | 100 | - | 学生的名 |
| last_name | string | 是 | 1 | 100 | - | 学生的姓 |
| email | string (email) | 是 | - | - | RFC 5322 | 有效的电子邮件地址 |
| phone | string | 否 | - | 15 | `^\+?[1-9]\d{9,14}$` | E.164格式 |
| date_of_birth | datetime | 是 | - | - | ISO 8601 | 出生日期 |
| grade_level | int | 是 | 1 | 12 | - | 年级（K-12） |

质量检查表

• [ ] 所有外部输入都已验证：每个API输入都有模式
• [ ] 有帮助的错误消息：每个字段都有清晰、可操作的验证消息
• [ ] 不信任客户端验证：后端重新验证所有输入
• [ ] 长度/格式约束已定义：所有字符串字段都有最小/最大长度
• [ ] 分离输入/输出模型：创建/更新（输入）与输出（响应）
• [ ] 类型安全管道：前端Zod推断TypeScript类型
• [ ] 应用清理：在验证前进行输入清理

集成点

技能	集成
@api-route-design	在response_model, Body()中使用模式
@error-handling	验证错误返回适当的错误代码
@fastapi-app	从Pydantic自动生成OpenAPI文档
@sqlmodel-crud	创建/更新模式与模型字段匹配