Featured image of post 从 Pydantic 到 msgspec: 一个关于数据校验与掌控的迁移故事
技术 Web开发

从 Pydantic 到 msgspec: 一个关于数据校验与掌控的迁移故事

FastAPI 迁移到 Litestar 的那些坑与顿悟——原来数据校验也可以不靠“魔法”一样强大

这次框架迁移, 让我怀疑人生

说到 Python 的数据校验, 我一度觉得自己稳如老狗。Pydantic 是我多年的好伙伴——优雅、强大, 还很贴心。一直以来, 我都以为世界上只有 Pydantic 一个数据校验“真命天子”, 谁还需要别的?

直到有一天, 我决定把项目从 FastAPI 迁移到 Litestar。理由很充分: 性能更好、架构更清晰、SQLAlchemy 深度集成……嗯, 听起来都很香。可惜天有不测风云, Litestar 推荐用 msgspec, 而不是 Pydantic。msgspec?说实话, 我只知道这个名字。

“切, 校验还不都一样?能有多难?”——我天真地想。

事实证明, 我错得离谱。

Pydantic 舒适区: 一把梭的全能数据校验

来, 给大家看看我之前写 Pydantic 校验的神仙写法:

from pydantic import BaseModel, EmailStr, field_validator

class UserRegistration(BaseModel):
    email: EmailStr  # 自动邮箱格式校验
    password: str
    age: int

    @field_validator('password')
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError('密码至少8位')
        if not any(c.isupper() for c in v):
            raise ValueError('密码需要包含大写字母')
        return v

    @field_validator('age')
    @classmethod
    def validate_age(cls, v: int) -> int:
        if v < 18:
            raise ValueError('必须年满18岁')
        return v

# FastAPI 用法
@app.post("/register")
async def register(data: UserRegistration):
    # 走到这里说明 data 已经被校验得明明白白
    user = await create_user(data.model_dump())
    return {"message": "用户已创建"}

一份模型定义, 所有校验规则都在里面。数据只要能变成 UserRegistration 实例, 类型、格式、业务规则全都通过。用惯了 Pydantic, 真有点“人生苦短, 我用 Pydantic”的豪气。

初见 msgspec: 我的校验器去哪儿了?

到了 Litestar, 我开始把 Pydantic 模型改成 msgspec 版本:

import msgspec

class UserRegistration(msgspec.Struct):
    email: str
    password: str
    age: int

@post("/register")
async def register(data: UserRegistration) -> dict:
    # 怎么校验邮箱格式?密码强度去哪儿写?
    # 年龄 < 18 怎么拦?
    user = await create_user(msgspec.to_builtins(data))
    return {"message": "用户已创建"}

我盯着屏幕陷入沉思。没有装饰器?没有 EmailStr?自定义校验咋整?疯狂 Google “msgspec field_validator 怎么用”, 一无所获。

我第一反应: “这库不行吧, 怎么用啊?”

第二反应: “难道是我没领会精髓?”

没错, 我确实漏掉了点什么。

真相只有一个: 校验其实有两种流派

折腾了好几个小时, 啃了一堆文档, 终于恍然大悟: Pydantic 和 msgspec 对“数据校验”这件事, 理解完全不一样!

来, 划重点:

Pydantic 哲学: “一锤子买卖, 模型边界一切校验”

Pydantic 主张: 只要创建模型实例, 所有校验一锅端, 类型、格式、业务规则、约束全都来。模型就是真理唯一源泉。

# Pydantic: 一次校验, 全部搞定
user = UserRegistration(**incoming_data)
# ↑ 只要这行没抛错, 类型、邮箱、密码、年龄都OK

msgspec 哲学: “类型校验飞快, 业务校验你自便”

msgspec 则主张分而治之:

  • 类型校验: 自动, 极快, 像闪电一样
  • 业务校验: 自己写, 自己掌控, 自己负责
# msgspec: 两步走
user = UserRegistration(**incoming_data)  # 类型校验, 快!
# ↓ 业务规则校验, 自己补
if len(user.password) < 8:
    raise ValueError("密码太短了")

刚开始我觉得这有点反人类, 干嘛要拆开?Pydantic 一步到位它不香吗?

后来我发现, 这其实很妙。

灵光一现: 校验的主动权回到我手里

转折点来了: msgspec 把业务校验的“开关”交给了你!

举个实际的例子:

密码重置的尴尬

我的应用里, 注册和重置密码是两个接口, 校验逻辑差不多但略有不同。

Pydantic 的写法:

class UserRegistration(BaseModel):
    email: EmailStr
    password: str
    full_name: str

    @field_validator('password')
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError('密码至少8位')
        return v

class PasswordReset(BaseModel):
    email: EmailStr
    password: str
    reset_token: str

    @field_validator('password')
    @classmethod
    def validate_password(cls, v: str) -> str:
        # 又写一遍……
        if len(v) < 8:
            raise ValueError('密码至少8位')
        return v

看出来没?密码校验代码得写两遍, 业务需求一变, 两个地方都得改, 心累。

msgspec 的玩法:

class UserRegistration(msgspec.Struct):
    email: str
    password: str
    full_name: str

class PasswordReset(msgspec.Struct):
    email: str
    password: str
    reset_token: str

# 校验逻辑集中在服务层
class UserService:
    @staticmethod
    def validate_password(password: str) -> None:
        if len(password) < 8:
            raise ValueError("密码至少8位")
        if not any(c.isupper() for c in password):
            raise ValueError("密码需包含大写字母")
        if not any(c.isdigit() for c in password):
            raise ValueError("密码需包含数字")

    async def register(self, data: UserRegistration):
        self.validate_password(data.password)
        # ... 创建用户

    async def reset_password(self, data: PasswordReset):
        self.validate_password(data.password)
        # ... 重置密码

校验方法只写一份, 两处都能用, 业务变更只改一地, 省心多了。

其实 msgspec Struct 也能写方法

我最初以为 msgspec Struct 就是个“数据罐头”, 其实你可以给它加自定义方法!

class UserRegistration(msgspec.Struct):
    email: str
    password: str
    age: int

    def validate(self) -> None:
        # 邮箱格式校验
        if "@" not in self.email or "." not in self.email.split("@")[-1]:
            raise ValueError("邮箱格式不对")
        # 密码校验
        if len(self.password) < 8:
            raise ValueError("密码至少8位")
        if not any(c.isupper() for c in self.password):
            raise ValueError("密码需包含大写字母")
        # 年龄校验
        if self.age < 18:
            raise ValueError("必须年满18岁")

@post("/register")
async def register(data: UserRegistration) -> dict:
    # 类型校验 msgspec 自动完成
    # 业务校验, 我说了算
    data.validate()
    user = await create_user(msgspec.to_builtins(data))
    return {"message": "用户已创建"}

这样你想“模型内校验”也没问题, 只不过需要你主动调用。这点Pydantic是自动的, msgspec给了你选择权:

  • 你可以在请求生命周期的任意阶段校验
  • 内部调用可以选择跳过校验
  • 校验规则可以根据上下文定制
  • 可以把校验错误和反序列化错误区分开

性能惊喜: msgspec 真的快到离谱

本来我没想过性能问题, 纯粹是想让校验别掉链子。结果跑了一下 benchmark, 吓一跳:

import msgspec
import pydantic
import time

class PydanticUser(pydantic.BaseModel):
    id: int
    email: str
    full_name: str
    is_active: bool

class MsgspecUser(msgspec.Struct):
    id: int
    email: str
    full_name: str
    is_active: bool

data = {
    "id": 12345,
    "email": "[email protected]",
    "full_name": "John Doe",
    "is_active": True
}

# Pydantic
start = time.perf_counter()
for _ in range(100_000):
    user = PydanticUser(**data)
pydantic_time = time.perf_counter() - start

# msgspec
start = time.perf_counter()
for _ in range(100_000):
    user = msgspec.convert(data, type=MsgspecUser)
msgspec_time = time.perf_counter() - start

print(f"Pydantic: {pydantic_time:.3f}s")
print(f"msgspec:  {msgspec_time:.3f}s")
print(f"msgspec快了 {pydantic_time / msgspec_time:.1f} 倍")

# 输出 (Ubuntu 24.04.3 LTS, Python 3.14.0b2) :
# Pydantic: 0.053s
# msgspec:  0.017s
# msgspec快了3.2倍

3.2倍! 只做类型校验就能快成这样。API 响应立马顺滑不少, 压力大时优势更明显。

实战对比: 一图胜千言

用户注册场景, 邮箱、密码、年龄都要校验。

Pydantic 方案

from pydantic import BaseModel, EmailStr, field_validator

class UserRegistration(BaseModel):
    email: EmailStr
    password: str
    age: int
    full_name: str | None = None

    @field_validator('password')
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError('密码至少8位')
        if not any(c.isupper() for c in v):
            raise ValueError('密码需包含大写字母')
        if not any(c.isdigit() for c in v):
            raise ValueError('密码需包含数字')
        return v

    @field_validator('age')
    @classmethod
    def validate_age(cls, v: int) -> int:
        if v < 18:
            raise ValueError('必须年满18岁')
        if v > 120:
            raise ValueError('年龄不合理')
        return v

@app.post("/register")
async def register(data: UserRegistration):
    user = await user_service.create(data.model_dump())
    return {"id": user.id}

优点:

  • 一处定义, 全部校验
  • 实例化自动校验
  • EmailStr 自带邮箱格式校验
  • 代码干净

缺点:

  • 校验与请求模型强绑定
  • 校验逻辑难以复用
  • 内部调用想跳过校验不方便
  • 序列化速度一般

msgspec 方案

import msgspec

class UserRegistration(msgspec.Struct):
    email: str
    password: str
    age: int
    full_name: str | None = None

    def validate(self) -> None:
        if "@" not in self.email or "." not in self.email.split("@")[-1]:
            raise ValueError("邮箱格式不对")
        if len(self.password) < 8:
            raise ValueError("密码至少8位")
        if not any(c.isupper() for c in self.password):
            raise ValueError("密码需包含大写字母")
        if not any(c.isdigit() for c in self.password):
            raise ValueError("密码需包含数字")
        if self.age < 18:
            raise ValueError("必须年满18岁")
        if self.age > 120:
            raise ValueError("年龄不合理")

@post("/register")
async def register(data: UserRegistration) -> dict:
    data.validate()
    user = await user_service.create(msgspec.to_builtins(data))
    return {"id": user.id}

优点:

  • 类型校验飞快
  • 校验时机你说了算
  • 校验逻辑可复用
  • 需要时可以跳过校验
  • 关注点分离更清晰

缺点:

  • 需要自己记得调用 .validate()
  • 没有现成的 EmailStr, 需要自己写格式校验
  • 校验代码手动敲, 略多一点
  • 控制器代码多一行

校验逻辑集中到服务层

msgspec 也可以把业务校验搬到 service:

class UserRegistration(msgspec.Struct):
    email: str
    password: str
    age: int
    full_name: str | None = None

class UserService:
    async def create(self, data: dict) -> User:
        if "@" not in data["email"]:
            raise ValueError("邮箱格式不对")
        if await self.repository.email_exists(data["email"]):
            raise ValueError("邮箱已被注册")
        self._validate_password(data["password"])
        if data["age"] < 18:
            raise ValueError("必须年满18岁")
        return await self.repository.create(data)

    @staticmethod
    def _validate_password(password: str) -> None:
        if len(password) < 8:
            raise ValueError("密码太短")
        # ... 其他校验

@post("/register")
async def register(data: UserRegistration) -> dict:
    user = await user_service.create(msgspec.to_builtins(data))
    return {"id": user.id}

这样业务逻辑和模型彻底解耦, 代码更易维护。

何时用谁?老司机经验分享

用过两套方案后, 我的结论如下:

适合用 Pydantic 的场景

  • 想要数据校验全自动: 写好模型, 啥都不用管
  • 标准 CRUD 接口: Pydantic 模式成熟
  • 团队偏爱“魔法”: 少写代码, 更自动
  • 需要生态支持: Pydantic 兼容无数第三方库
  • 性能不是核心诉求: 够用就行

适合用 msgspec 的场景

  • 极致性能要求: 高并发、实时系统首选
  • 需要校验主动权: 你决定何时校验
  • 业务逻辑要复用、要解耦: 校验代码集中管理
  • 用 Litestar 框架: 官方推荐, 效率最佳
  • 想要关注点分离: service 层搞定业务校验

还能混着用

小秘密: 其实两者可以共存。复杂操作用 Pydantic, 性能瓶颈用 msgspec。

# 管理后台复杂校验用 Pydantic
class ComplexUserUpdate(BaseModel):
    ...

# 高频接口数据返回用 msgspec
class UserListResponse(msgspec.Struct):
    ...

总结: 校验没有唯一正确的姿势

这次迁移让我明白了: **没有“唯一正确”的数据校验方式。**Pydantic 和 msgspec 只是不同的哲学:

  • Pydantic: 校验应该自动, 且面面俱到
  • msgspec: 校验应该显式, 且极致高效

各有优劣, 视场景选用。

刚开始我觉得 msgspec “不完整”, 后来发现显式的校验反而让代码更清晰、好维护.validate() 这行代码, 从“啰嗦”变成了“注释”, 明确告诉你“这里发生了业务校验”。手写校验方法还能独立测试和复用。

至于性能提升?那纯属意外之喜。

迁移还在继续

我的 Litestar 迁移之路还没走完, 但我已经不再纠结 msgspec 的“与众不同”。曾经的迷惑, 成了新的认可。

我会全盘回到 Pydantic 吗?不会。

我会只用 msgspec 吗?也不会。

关键在于: **理解每个工具的设计哲学。**Pydantic 不是“全能”, msgspec 也不是“残缺”——只是各自的优化点不同。用对了场景, 才是真本事。

有时候, 那些你以为“少了点啥”的库, 其实给了你更多自由和掌控感。


(本篇由人类原创, 部分细节参考 AI 建议优化 😃)

© 2022 - 2026 张欣耕

保留所有权利