如何在Pydantic中定义bson.binary.Binary字段适配MongoDB

Pydantic无法自动为bson.binary.Binary生成JSON Schema，导致OpenAPI文档加载失败。以下是三种可行的解决方法：

方案一：自定义Pydantic类型继承bson.Binary #

通过自定义类型，让Pydantic明确该类型的序列化/反序列化规则及对应的JSON Schema：

from bson.binary import Binary
from pydantic_core import core_schema
from pydantic import GetCoreSchemaHandler
from typing import Any
import base64

class PydanticBinary(Binary):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        # 序列化：将Binary转为base64字符串
        def serialize(value: Binary) -> str:
            return value.base64()
        
        # 反序列化：将base64字符串转为Binary
        def deserialize(value: str) -> Binary:
            return Binary(base64.b64decode(value))
        
        # 定义核心schema：支持直接传入Binary或字符串，序列化输出字符串
        return core_schema.no_info_after_validator_function(
            deserialize,
            core_schema.union_schema([
                core_schema.is_instance_schema(Binary),
                core_schema.str_schema()
            ]),
            serialization=core_schema.plain_serializer_function_ser_schema(serialize)
        )

然后在模型中替换原Binary类型：

class Form(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    is_attachment: bool = Field(...)
    content_type: str = Field(...)
    file_name: str = Field(...)
    file_data: PydanticBinary = Field(...)
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra={
            "example": {
                "file_name": "file.pdf",
                "is_attachment": True,
                "content_type": "application/pdf",
                "file_data": "JVBERi0xLjQK...",
            }
        },
    )

方案二：用Annotated结合验证器与序列化器 #

无需自定义类型，直接通过Annotated为字段添加序列化和反序列化逻辑：

from bson.binary import Binary
from pydantic import BeforeValidator, PlainSerializer
from typing import Annotated
import base64

# 定义带处理逻辑的Binary类型别名
BinaryStr = Annotated[
    Binary,
    # 反序列化：字符串转Binary
    BeforeValidator(lambda v: Binary(base64.b64decode(v)) if isinstance(v, str) else v),
    # 序列化：Binary转base64字符串
    PlainSerializer(lambda v: v.base64(), return_type=str)
]

在模型中使用这个别名：

class Form(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    is_attachment: bool = Field(...)
    content_type: str = Field(...)
    file_name: str = Field(...)
    file_data: BinaryStr = Field(...)
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        json_schema_extra={
            "example": {
                "file_name": "file.pdf",
                "is_attachment": True,
                "content_type": "application/pdf",
                "file_data": "JVBERi0xLjQK...",
            }
        },
    )

方案三：接口返回时手动转换数据 #

如果不想修改模型，可以在接口查询MongoDB后，手动将Binary字段转为base64字符串：

@app.get("/forms/", response_description="List all forms", response_model=FormCollection, response_model_by_alias=False,)
async def list_forms():
    forms = await forms_collection.find().to_list(1000)
    # 遍历转换每个文档的file_data字段
    for form in forms:
        if isinstance(form.get("file_data"), Binary):
            form["file_data"] = form["file_data"].base64()
    return FormCollection(forms=forms)

注意事项 #

• 三种方案的核心都是将bson.Binary与JSON兼容的字符串（base64）进行双向转换，确保Pydantic能生成正确的JSON Schema。
• 如果涉及文件上传，可通过FastAPI的UploadFile接收文件，再将其内容转为Binary存储到MongoDB。

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