MongoDB 二进制字段的 Pydantic 自定义类型

Question

我需要从包含图像字段的 mongodb 集合中检索记录列表，并将它们转换为 Pydantic 模型。

Pydantic中有对应的bson.Binary类型吗？或者一种将二进制数据转换为 Pydantic 可以验证的类型的方法？

我尝试过“字节”：

class Category(BaseModel):
    category_name: str = Field(...)
    category_icon_binary: Optional[bytes] = Field(...)

但我得到：

UnicodeDecodeError: 'utf-8' codec can't decode byte 0x89 in position 0: invalid utf-8

我尝试过“bson.Binary”和任意类型_允许：

class Category(BaseModel):
    category_name: str = Field(...)
    category_icon_binary: Optional[Binary] = Field(...)
    model_config = ConfigDict(arbitrary_types_allowed=True,....)

我得到：

  Input should be an instance of Binary [type=is_instance_of, input_value=b'\x89PNG\r\n\x1a\n\x00\x...0\x00\x00IEND\xaeB`\x82', input_type=bytes]

这是包含字节字段（或二进制）的模型：

# Pydantic model

from pydantic import BaseModel, Field, BeforeValidator
from typing import Optional, Annotated, List
from dataclasses import dataclass

PyObjectId = Annotated[str, BeforeValidator(str)]

class Category(BaseModel):
    category_id: Optional[PyObjectId] = Field(alias="_id", default=None)
    category_name: str = Field(...)
    category_icon_binary: Optional[bytes] = Field(...)
    content_type: Optional[str] = Field(...)

class CategoriesCollection(BaseModel):
    categories: List[Category]

这是使用模型的端点：

# API endpoint

@router.get("/", response_model=list[Category], response_model_by_alias=False,)
def get():
# THIS WORKS FINE there is no validation error in loading bytes in the model list[Category]
    result = CategoriesCollection(categories=categories_collection.find())

# THIS RAISES validation error UnicodeDecodeError
    return result.categories

这表明在返回之前没有验证错误并且二进制文件已正确加载：

这显示了文档如何保存在集合中：

Answer 1

根据添加的代码和屏幕截图，问题不在于 Pydantic，也不完全是。正如您所指出的，

Category

和

CategoriesCollection

的 Pydantic 模型正在正确加载，没有错误。

问题来自

response_model

，因为您尝试以 JSON 格式返回二进制数据，这是不允许的 - 而且 FastAPI/Pydantic 没有为此提供默认处理程序，与 datettime、UUID 等不同.

您需要使用 Base64 编码或 Base85 等将其转换为 JSON 可接受的字符串，其中

@field_serializer

:

import base64

class Category(BaseModel):
    category_id: Optional[PyObjectId] = Field(alias="_id", default=None)
    category_name: str = Field(...)
    category_icon_binary: Optional[bytes] = Field(...)
    content_type: Optional[str] = Field(...)
    
    @field_serializer('category_icon_binary', when_used='json-unless-none')
    def bytes_as_base64(b: bytes):
        return base64.b64encode(b)

注意：由于字节与字符串，响应处理程序可能仍然存在问题。所以更新模型有

category_icon_binary: Optional[Union[bytes, str]] = Field(...)

---

或者，您可以创建一个类似

OutputCategory

的类，它继承自

Category

并覆盖

category_icon_binary

并将其转换为 Base64、十六进制或您选择的任何内容。就像 In/OutUser 的 FastAPI 文档示例

class OutputCategory(Category):  # inherit from Category
    category_icon_binary: Optional[str] = Field(...)  # note the change in type
    
   # and add a validator which will convert the bin to hex/base64
   ...