基于Beanie ODM、FastAPI与MongoDB优化JSON响应引用ID格式的问题

直接通过自定义Beanie字段和Link子类，把序列化逻辑嵌入底层，避免手动遍历格式化的性能损耗，具体实现步骤如下：

1. 自定义带前缀的主键字段 #

针对自身_id的booking_xxx格式需求，继承PydanticObjectId实现带前缀的序列化：

from beanie import PydanticObjectId
from pydantic_core import core_schema

class PrefixedObjectId(PydanticObjectId):
    prefix: str
    
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler):
        base_schema = super().__get_pydantic_core_schema__(source_type, handler)
        return core_schema.no_info_wrap_validator_function(
            cls._serialize,
            base_schema,
            serialization=core_schema.plain_serializer_function_ser_schema(cls._serialize)
        )
    
    @classmethod
    def _serialize(cls, value: PydanticObjectId) -> str:
        return f"{cls.prefix}_{str(value)}"

# 针对Booking模型的专属ID类
class BookingId(PrefixedObjectId):
    prefix = "booking"

2. 自定义带集合前缀的Link字段 #

处理引用字段的account_xxx格式需求，继承Beanie的Link类重写序列化逻辑：

from beanie import Link
from pydantic import GetCoreSchemaHandler
from pydantic_core import core_schema

class PrefixedLink(Link):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        base_schema = super().__get_pydantic_core_schema__(source_type, handler)
        return core_schema.no_info_wrap_validator_function(
            cls._serialize_link,
            base_schema,
            serialization=core_schema.plain_serializer_function_ser_schema(cls._serialize_link)
        )
    
    @classmethod
    def _serialize_link(cls, value) -> str:
        # 直接从引用文档中取集合名和ID拼接
        return f"{value.__collection_name__}_{str(value.id)}"

3. 在Booking模型中应用自定义字段 #

把自定义字段替换原生字段，所有序列化逻辑自动生效：

from beanie import Document
from pydantic import Field
from models.account import Account  # 你的Account模型

class Booking(Document):
    id: BookingId = Field(default_factory=BookingId, alias="_id")
    account: PrefixedLink[Account]
    # 其他业务字段...
    
    class Settings:
        name = "bookings"

4. FastAPI中直接返回批量数据 #

现在查询批量数据时，不需要额外格式化，直接返回即可，性能和原生序列化一致：

from fastapi import FastAPI
from models.booking import Booking

app = FastAPI()

@app.get("/bookings")
async def get_all_bookings():
    # 直接返回10000条格式化后的结果
    return await Booking.find_all().to_list(10000)

内容的提问来源于stack exchange，提问作者Daniel Alvarado