FastAPI+Tortoise-ORM下自定义递归关联Pydantic模型问题咨询

核心结论 #

Tortoise-ORM 完全支持脱离 pydantic_model_creator 单独和自定义 Pydantic 模型配合使用，二者没有强制绑定关系，你之前基于 SQLAlchemy 的分层逻辑几乎可以直接复用。

具体实现方案 #

你原有自定义的 Pydantic 模型不需要大改，只需要做少量适配即可：

1. 给自定义 Pydantic 模型开启 ORM 适配配置
2. 查询 Tortoise 模型时提前拉取需要的关联字段和聚合字段
3. 用 Pydantic 自带的 from_orm 方法直接转换 Tortoise ORM 实例

第一步：定义Tortoise数据库模型

from tortoise.models import Model
from tortoise import fields

class Domain(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255)
    
    class Meta:
        table = "domain"

class Mailbox(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=255)
    # 外键会自动生成domain_id属性，不需要额外定义
    domain = fields.ForeignKeyField("models.Domain", related_name="mailboxes", on_delete=fields.CASCADE)
    
    class Meta:
        table = "mailbox"

第二步：适配自定义Pydantic模型

from pydantic import BaseModel
from typing import List

class DomainBase(BaseModel):
    id: int
    name: str

    class Config:
        # Pydantic V2 写法，V1版本替换为 orm_mode = True
        from_attributes = True

class Domain(DomainBase):
    mailboxes: List["MailboxSummary"] = []

class DomainSummary(DomainBase):
    mailboxes_count: int

class MailboxBase(BaseModel):
    id: int
    username: str

    class Config:
        from_attributes = True

class Mailbox(MailboxBase):
    domain: DomainSummary

class MailboxSummary(MailboxBase):
    domain_id: int

# 处理循环引用
Domain.model_rebuild()
Mailbox.model_rebuild()

第三步：查询和转换示例

• 查询带关联邮箱列表的Domain

async def get_domain(domain_id: int):
    # 提前预加载关联的mailboxes数据，避免懒加载报错
    domain_obj = await Domain.get_or_none(id=domain_id).prefetch_related("mailboxes")
    if not domain_obj:
        return None
    # 直接转换为自定义Pydantic模型
    return Domain.from_orm(domain_obj)

• 查询带邮箱数量的DomainSummary

from tortoise.functions import Count

async def get_domain_summary(domain_id: int):
    # 用annotate聚合计算邮箱数量，字段直接挂载到ORM实例
    domain_obj = await Domain.annotate(
        mailboxes_count=Count("mailboxes")
    ).get_or_none(id=domain_id)
    return DomainSummary.from_orm(domain_obj)

• 查询带所属域信息的Mailbox

async def get_mailbox(mailbox_id: int):
    mailbox_obj = await Mailbox.get_or_none(id=mailbox_id).prefetch_related("domain")
    if not mailbox_obj:
        return None
    return Mailbox.from_orm(mailbox_obj)

补充说明 #

pydantic_model_creator 本质只是辅助生成Pydantic模型的脚手架工具，没有任何运行时依赖，完全可以弃用。你完全可以保持数据库层和表现层的独立结构，不会出现二者耦合的问题。

内容的提问来源于stack exchange，提问作者Nathan Ward