使用Beanie+FastAPI构建嵌套分类时BackLink序列化报错如何解决？

错误根源 #

你遇到的Unable to serialize unknown type: <class 'beanie.odm.fields.BackLink'>错误，核心原因是BackLink是Beanie提供的动态反向查询字段，并非存储在数据库的实体字段，它无法被序列化或手动赋值。你之前的代码试图将children（BackLink类型）直接赋值为Category对象列表，触发了序列化失败。

同时你的模型结构存在继承冲突：CategoryBase被定义为Document，导致后续CategoryIn和Category的字段继承出现混淆，进一步加剧了序列化问题。

修正步骤 #

1. 重构模型结构 #

将基类改为BaseModel，明确区分输入模型与数据库模型的职责：

from pydantic import Field, BaseModel
from typing import Optional, Annotated, List
from beanie import Document, Link, BackLink, PydanticObjectId, WriteRules, Indexed
from datetime import datetime, timezone

# 基础数据结构，仅定义业务字段
class CategoryBase(BaseModel):
    name: Annotated[str, Indexed(unique=True)] = Field(max_length=100)
    description: Optional[str] = Field(max_length=800, default=None)
    options: Optional[List[str]] = Field(default_factory=list)

# 前端输入模型，仅接收ID类型的关联字段
class CategoryIn(CategoryBase):
    parent: Optional[PydanticObjectId] = None
    children: Optional[List[PydanticObjectId]] = []

# 数据库映射模型，继承Document与业务基类
class Category(Document, CategoryBase):
    parent: Optional[Link["Category"]] = None
    # BackLink仅用于反向查询，无需手动赋值或存储
    children: Optional[List[BackLink["Category"]]] = Field(original_field="parent", default_factory=list)
    date_created: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    date_modified: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    
    class Settings:
        name = "categories"

# 解决循环类型引用
Category.update_forward_refs()

2. 重写创建逻辑 #

移除对BackLink字段的手动赋值，利用Beanie的Link机制和WriteRules维护关联关系：

class CategoryOps:
    def __init__(self, _id: PydanticObjectId):
        self.id = _id
    
    async def get_category(self) -> Optional[Category]:
        return await Category.get(self.id)

class CategoryRepo:
    @staticmethod
    async def create(cat_in: CategoryIn) -> Category:
        # 排除输入模型中的children字段，避免传入BackLink导致序列化错误
        category_data = cat_in.dict(exclude={"children"})
        
        # 处理父分类关联：将ID转为Link对象
        if cat_in.parent:
            parent_category = await CategoryOps(cat_in.parent).get_category()
            if parent_category:
                category_data["parent"] = parent_category
        
        # 创建并保存新分类，WriteRules.WRITE自动维护关联关系
        new_category = Category(**category_data)
        await new_category.save(link_rule=WriteRules.WRITE)
        
        # 处理子分类关联：手动更新子分类的parent字段
        if cat_in.children:
            for child_id in cat_in.children:
                child_category = await CategoryOps(child_id).get_category()
                if child_category:
                    child_category.parent = new_category
                    await child_category.save(link_rule=WriteRules.WRITE)
        
        # 同步关联数据，确保反向查询生效
        await new_category.sync()
        return new_category

关键说明 #

1. BackLink的本质：它是Beanie基于original_field（这里是parent）实现的动态查询接口，不会存储在数据库中，因此永远不需要手动赋值。当子分类的parent指向父分类时，父分类的children字段会自动查询到所有关联的子分类。
2. 关联维护逻辑：
   • 父分类关联：通过将新分类的parent设为父分类的Link对象，保存时用WriteRules.WRITE，Beanie会自动维护关联关系。
   • 子分类关联：需要手动遍历子分类，将它们的parent更新为新分类的Link，再保存子分类，这样新分类的children反向查询就能获取到这些子分类。
3. 模型分离：输入模型（CategoryIn）仅负责接收前端传入的ID，数据库模型（Category）负责映射MongoDB文档，避免字段类型冲突。

内容的提问来源于stack exchange，提问作者ai.devmir