FastAPI结合MongoDB时如何在Schema中使用Decimal类型？

问题描述 #

我开发了一个基于Python FastAPI和MongoDB的应用，为避免float类型的浮点精度问题，希望在Schema中使用Decimal类型（比如rating字段，后续还会用于price字段），但使用condecimal时向MongoDB插入文档出现BSON编码错误。

Schema定义

class FlowerTypeSchema(BaseModel):
    id: UUID = None
    merchant_id: Optional[UUID] = None
    name: LangSchema
    description: LangSchema
    photo: Optional[str] = None
    custom: Optional[bool] = False
    rating: Optional[condecimal(ge=0)] = 0
    tags: List[str]
    consumables: List[ProductsSpentSchema]
    created_at: datetime = None
    updated_at: datetime = None
    deleted_at: datetime = None

创建文档代码

async def create_flower_type(flower_type: FlowerTypeSchema) -> FlowerTypeResponse:
    flower_type.id = uuid4()
    flower_type.photo = None
    flower_type.created_at = datetime.now(pytz.timezone(TIMEZONE)).strftime("%Y-%m-%dT%H:%M:%S")
    flower_type.updated_at = datetime.now(pytz.timezone(TIMEZONE)).strftime("%Y-%m-%dT%H:%M:%S")
    await flower_types_collection.insert_one(flower_type.dict())
    return FlowerTypeResponse(**flower_type.dict())

报错信息

2024-06-01 14:46:32,147 - uvicorn.error - ERROR - Exception in ASGI application
Traceback (most recent call last):
  # 省略中间栈信息
bson.errors.InvalidDocument: cannot encode object: Decimal('4.5'), of type: <class 'decimal.Decimal'>

问题原因 #

MongoDB的BSON格式不直接支持Python标准库的decimal.Decimal类型，必须转换为MongoDB原生支持的bson.Decimal128类型才能正常插入。

解决方法 #

方法一：插入前手动转换字段类型

在将数据插入MongoDB前，手动把rating字段的Decimal值转换为Decimal128：

from bson import Decimal128
# 其他导入保持不变

async def create_flower_type(flower_type: FlowerTypeSchema) -> FlowerTypeResponse:
    flower_type.id = uuid4()
    flower_type.photo = None
    now = datetime.now(pytz.timezone(TIMEZONE)).strftime("%Y-%m-%dT%H:%M:%S")
    flower_type.created_at = now
    flower_type.updated_at = now
    
    # 转换Decimal为MongoDB支持的Decimal128
    flower_data = flower_type.dict()
    if flower_data.get("rating") is not None:
        # 转成字符串再初始化Decimal128，避免精度丢失
        flower_data["rating"] = Decimal128(str(flower_data["rating"]))
    
    await flower_types_collection.insert_one(flower_data)
    return FlowerTypeResponse(**flower_type.dict())

方法二：用Pydantic字段序列化器自动转换

在Schema中添加序列化器，让Pydantic在生成字典时自动把Decimal转成Decimal128：

from bson import Decimal128
from pydantic import field_serializer
# 其他导入保持不变

class FlowerTypeSchema(BaseModel):
    id: UUID = None
    merchant_id: Optional[UUID] = None
    name: LangSchema
    description: LangSchema
    photo: Optional[str] = None
    custom: Optional[bool] = False
    rating: Optional[condecimal(ge=0)] = 0
    tags: List[str]
    consumables: List[ProductsSpentSchema]
    created_at: datetime = None
    updated_at: datetime = None
    deleted_at: datetime = None

    @field_serializer('rating')
    def serialize_rating(self, value: Optional[Decimal], _info):
        if value is not None:
            return Decimal128(str(value))
        return value

查询数据的反向转换 #

从MongoDB查询数据时，返回的是Decimal128类型，需要转换回Decimal才能在Pydantic响应模型中正常解析，可添加字段验证器：

from pydantic import field_validator
# 其他导入保持不变

class FlowerTypeResponse(BaseModel):
    # 保持原有字段定义
    id: UUID
    merchant_id: Optional[UUID]
    name: LangSchema
    description: LangSchema
    photo: Optional[str]
    custom: Optional[bool]
    rating: Optional[condecimal(ge=0)] = 0
    tags: List[str]
    consumables: List[ProductsSpentSchema]
    created_at: str
    updated_at: str
    deleted_at: Optional[str]

    @field_validator('rating', mode='before')
    def parse_rating(cls, value):
        if isinstance(value, Decimal128):
            return Decimal(str(value))
        return value

内容的提问来源于stack exchange，提问作者Акмаль