pydantic_model_creator 完整用法

官方文档：https://tortoise.github.io/contrib/pydantic.html
版本：tortoise-orm >= 0.20，pydantic >= 2.0
最后更新：2026-04-11

一、函数签名与参数

from tortoise.contrib.pydantic import pydantic_model_creator

UserSchema = pydantic_model_creator(User, name="UserSchema")

pydantic_model_creator 参数表

二、基础用法

2.1 模型定义

# models.py
from tortoise import fields
from tortoise.models import Model


class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=255)
    password_hash = fields.CharField(max_length=128)
    is_active = fields.BooleanField(default=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    posts: fields.ReverseRelation["Post"]

    class Meta:
        table = "user"

2.2 生成基础 Schema

from tortoise.contrib.pydantic import pydantic_model_creator

# 生成包含所有字段的 Schema
UserSchema = pydantic_model_creator(User, name="UserSchema")

# 排除敏感字段
UserOut = pydantic_model_creator(
    User,
    name="UserOut",
    exclude=("password_hash",)
)

# 只包含指定字段
UserSummary = pydantic_model_creator(
    User,
    name="UserSummary",
    include=("id", "username", "email")
)

2.3 从模型实例转换

# from_tortoise_orm：从单个模型实例生成 Pydantic 模型
user = await User.get(id=1)
user_data = await UserOut.from_tortoise_orm(user)

# from_queryset：从 QuerySet 生成 Pydantic 模型列表
users = await UserOut.from_queryset(User.all())

# from_queryset_single：从返回单对象的 QuerySet 生成（不需要先 await）
user_data = await UserOut.from_queryset_single(User.get(id=1))

三、分层 Schema 策略

不同操作场景需要不同的 Schema，推荐将 Create、Update、Response 分离定义。

from tortoise.contrib.pydantic import pydantic_model_creator

# Response Schema：返回给客户端，排除敏感字段
UserOut = pydantic_model_creator(
    User,
    name="UserOut",
    exclude=("password_hash",)
)

# Create Schema：创建时不需要只读字段（id、created_at、updated_at）
UserCreate = pydantic_model_creator(
    User,
    name="UserCreate",
    exclude_readonly=True,      # 排除 id、auto_now、auto_now_add 字段
    exclude=("is_active",)      # 额外排除不需要用户填写的字段
)

# Update Schema：更新时所有字段都是可选的
UserUpdate = pydantic_model_creator(
    User,
    name="UserUpdate",
    exclude_readonly=True,
    exclude=("is_active",),
    optional=("username", "email", "password_hash")  # 所有字段变为可选
)

四、关联关系序列化

当模型包含外键或反向关系时，pydantic_model_creator 会自动将关联对象嵌套序列化，但必须先 prefetch_related。

4.1 模型定义

class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    author: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
        "models.User", related_name="posts"
    )
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "post"

4.2 生成带嵌套关联的 Schema

# User Schema 包含嵌套的 posts 列表
UserWithPosts = pydantic_model_creator(
    User,
    name="UserWithPosts",
    exclude=("password_hash",)
)

# 必须 prefetch_related，否则关联字段报错
user = await User.get(id=1).prefetch_related("posts")
data = await UserWithPosts.from_tortoise_orm(user)
# data.posts 为 PostSchema 列表

# 排除嵌套关联中的特定字段（使用点号路径）
UserWithPostTitles = pydantic_model_creator(
    User,
    name="UserWithPostTitles",
    exclude=("password_hash", "posts.content", "posts.created_at")
)

4.3 Post Schema 包含作者信息

PostWithAuthor = pydantic_model_creator(
    Post,
    name="PostWithAuthor",
    exclude=("author.password_hash",)
)

post = await Post.get(id=1).prefetch_related("author")
data = await PostWithAuthor.from_tortoise_orm(post)

五、computed 计算字段

computed 参数可以将模型上的 @property 方法暴露为 Pydantic Schema 的字段。

class User(Model):
    id = fields.IntField(pk=True)
    first_name = fields.CharField(max_length=50)
    last_name = fields.CharField(max_length=50)
    email = fields.CharField(max_length=255)

    @property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

    @property
    def email_domain(self) -> str:
        return self.email.split("@")[-1]

    class Meta:
        table = "user"


# 包含计算属性
UserWithComputed = pydantic_model_creator(
    User,
    name="UserWithComputed",
    computed=("full_name", "email_domain")
)

user = await User.get(id=1)
data = await UserWithComputed.from_tortoise_orm(user)
print(data.full_name)      # "张 三"
print(data.email_domain)   # "example.com"

计算属性在 Pydantic Schema 中为只读字段（无法从外部赋值），类型根据 @property 的返回类型注解自动推断。

六、exclude_readonly 用于创建 Schema

exclude_readonly=True 会自动排除所有只读字段，包括：

• 主键字段（pk=True）
• auto_now=True 的 DatetimeField
• auto_now_add=True 的 DatetimeField
• generated=True 的字段

UserCreate = pydantic_model_creator(
    User,
    name="UserCreate",
    exclude_readonly=True
)

# 等价于手动 exclude=("id", "created_at", "updated_at")
# 但 exclude_readonly 更健壮，不需要手动维护字段列表

七、与 FastAPI 集成完整示例

# schemas.py
from tortoise.contrib.pydantic import pydantic_model_creator
from myapp.models import User, Post

UserOut = pydantic_model_creator(User, name="UserOut", exclude=("password_hash",))
UserCreate = pydantic_model_creator(User, name="UserCreate", exclude_readonly=True)
UserUpdate = pydantic_model_creator(
    User,
    name="UserUpdate",
    exclude_readonly=True,
    optional=("username", "email", "password_hash", "is_active")
)
PostOut = pydantic_model_creator(Post, name="PostOut")

# routers/users.py
from fastapi import APIRouter, HTTPException
from passlib.hash import bcrypt
from myapp.models import User
from myapp.schemas import UserOut, UserCreate, UserUpdate

router = APIRouter(prefix="/users", tags=["users"])


@router.get("/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return await UserOut.from_tortoise_orm(user)


@router.get("/", response_model=list[UserOut])
async def list_users():
    return await UserOut.from_queryset(User.filter(is_active=True))


@router.post("/", response_model=UserOut, status_code=201)
async def create_user(payload: UserCreate):
    # UserCreate 是 Pydantic 模型，用 .model_dump() 获取字典
    data = payload.model_dump()
    data["password_hash"] = bcrypt.hash(data.pop("password_hash"))
    user = await User.create(**data)
    return await UserOut.from_tortoise_orm(user)


@router.patch("/{user_id}", response_model=UserOut)
async def update_user(user_id: int, payload: UserUpdate):
    user = await User.get_or_none(id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    # exclude_unset=True：只更新客户端实际传入的字段
    update_data = payload.model_dump(exclude_unset=True)
    await user.update_from_dict(update_data).save()
    return await UserOut.from_tortoise_orm(user)


@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: int):
    deleted = await User.filter(id=user_id).delete()
    if not deleted:
        raise HTTPException(status_code=404, detail="User not found")

八、Pydantic v2 兼容性注意事项

tortoise-orm 0.20+ 已全面支持 Pydantic v2，以下是主要变化：

# Pydantic v2 用 model_dump() 代替 dict()
data = user_schema.model_dump()
data = user_schema.model_dump(exclude_unset=True)  # 只获取设置过的字段

# Pydantic v2 用 model_validate() 代替 parse_obj()
user_schema = UserCreate.model_validate({"username": "alice", "email": "[email protected]"})

# JSON 序列化用 model_dump_json()
json_str = user_schema.model_dump_json()

如果项目同时使用 Pydantic v1 和 v2（如通过兼容层），需要在配置中明确指定：

from pydantic import ConfigDict

UserOut = pydantic_model_creator(
    User,
    name="UserOut",
    model_config=ConfigDict(from_attributes=True)  # Pydantic v2 必须设置
)

九、踩坑与注意事项

9.1 循环引用问题

场景：User 有多个 Post，Post 又有 ForeignKey 指向 User，两者互相引用。

现象：生成 Schema 时无限嵌套，或运行时 RecursionError。

解决：设置 allow_cycles=True，Tortoise 会在第一次循环时截断展开：

UserWithPosts = pydantic_model_creator(
    User,
    name="UserWithPosts",
    allow_cycles=True,
    exclude=("password_hash",)
)

或者在一方的 Schema 中排除对方的关联字段：

PostOut = pydantic_model_creator(
    Post,
    name="PostOut",
    exclude=("author.posts",)  # 排除 author 中的 posts，避免循环
)

9.2 关联字段未 fetch 报错

现象：await UserOut.from_tortoise_orm(user) 时报 OperationalError: You need to fetch ... first。

原因：Schema 包含关联字段，但查询时没有预取关联数据。

解决：查询时用 prefetch_related 或 select_related 预取：

# 错误：缺少 prefetch_related
user = await User.get(id=1)
data = await UserWithPosts.from_tortoise_orm(user)  # 报错

# 正确
user = await User.get(id=1).prefetch_related("posts")
data = await UserWithPosts.from_tortoise_orm(user)  # 正常

# from_queryset 会自动处理预取，无需手动指定
users = await UserWithPosts.from_queryset(User.all())  # 自动 prefetch_related

预取关系详见 select_related 和 prefetch_related的区别和用法。

9.3 字段名与 Pydantic 保留字冲突

场景：模型字段名为 schema、model_fields 等 Pydantic 内部保留名称。

现象：生成的 Pydantic 模型行为异常或报错。

解决：修改模型字段名，或使用 include 参数只选择无冲突的字段。

9.4 name 参数不唯一导致 Schema 混用

pydantic_model_creator 内部会缓存相同 name 的 Schema。如果两处用相同 name 但不同参数调用，后者会复用前者的缓存，产生意料之外的结果。

规范：为每个 Schema 指定唯一的 name 参数。

# 错误：相同 name 不同 exclude
UserOut1 = pydantic_model_creator(User, name="User", exclude=("password_hash",))
UserOut2 = pydantic_model_creator(User, name="User", exclude=("email",))  # 实际是 UserOut1 的缓存

# 正确：使用不同名称
UserPublic = pydantic_model_creator(User, name="UserPublic", exclude=("password_hash",))
UserInternal = pydantic_model_creator(User, name="UserInternal", exclude=("email",))

最佳实践

每个 Schema 指定唯一 name：pydantic_model_creator 有内部缓存，相同 name 的调用返回第一次创建的结果。为每个变体传入独立 name（如 "UserPublic"、"UserCreate"），避免参数不同却拿到同一个 Schema。

用 exclude 代替手写 Schema 字段：对于只需隐藏少数字段（如 password_hash）的场景，pydantic_model_creator(User, exclude=("password_hash",)) 比手写 Pydantic 模型更安全，新增 ORM 字段时 Schema 自动跟进，不会遗漏。

需要关联数据时先 prefetch_related：pydantic_model_creator 生成的 Schema 序列化关联字段时，若关联未加载会抛出 NoValuesFetched。在调用 from_tortoise_orm 之前务必预加载：

user = await User.get(id=uid).prefetch_related('posts')
schema = await UserSchema.from_tortoise_orm(user)

输入 Schema 用 include 限制可写字段：创建/更新接口只允许写入特定字段时，用 include=("username", "email") 明确白名单，防止客户端传入 id、created_at 等只读字段被意外写入。

异步批量序列化用 from_queryset：列表接口用 await UserSchema.from_queryset(User.all()) 而非循环调用 from_tortoise_orm，减少异步调度开销，且内部自动处理关联预加载。

常见陷阱

陷阱：相同 name 的 Schema 行为不符合预期

现象： pydantic_model_creator(User, name="User", exclude=("email",)) 返回的 Schema 仍然包含 email 字段。
原因： 之前已有 pydantic_model_creator(User, name="User") 的调用，缓存中存了无 exclude 的版本，后续同名调用直接返回缓存。
解决： 为每个变体使用唯一 name；若需强制重建（测试中），调用 User._pydantic_cache.clear() 清空缓存。

陷阱：序列化时 NoValuesFetched 异常

现象： 调用 await UserWithPostsSchema.from_tortoise_orm(user) 时抛出 NoValuesFetched: posts。
原因： Schema 包含关联字段（posts），但 ORM 对象未预加载该关联。
解决： 查询时加上 prefetch_related('posts')：user = await User.get(id=uid).prefetch_related('posts')。

陷阱：computed 字段未出现在生成的 Schema 中

现象： 在 ORM 模型上用 @property 定义了计算属性，但 pydantic_model_creator 生成的 Schema 没有该字段。
原因： pydantic_model_creator 只识别 tortoise_orm 的字段描述符，Python 原生 @property 不在扫描范围内。
解决： 在模型的 PydanticMeta.computed 中声明计算属性名称：class PydanticMeta: computed = ("full_name",)，Tortoise 会在序列化时调用该属性。

参见

查询操作完全指南
初始化与配置
信号机制