MongoDB 的 FastAPI 问题 - TypeError：“ObjectId”对象不可迭代

Question

我在通过 FastAPI 插入 MongoDB 时遇到一些问题。

下面的代码按预期工作。请注意

response

变量尚未在

response_to_mongo()

中使用。

model

是一个sklearn ElasticNet模型。

app = FastAPI()


def response_to_mongo(r: dict):
    client = pymongo.MongoClient("mongodb://mongo:27017")
    db = client["models"]
    model_collection = db["example-model"]
    model_collection.insert_one(r)


@app.post("/predict")
async def predict_model(features: List[float]):

    prediction = model.predict(
        pd.DataFrame(
            [features],
            columns=model.feature_names_in_,
        )
    )

    response = {"predictions": prediction.tolist()}
    response_to_mongo(
        {"predictions": prediction.tolist()},
    )
    return response

但是，当我像这样编写

predict_model()

并将

response

变量传递给

response_to_mongo()

时：

@app.post("/predict")
async def predict_model(features: List[float]):

    prediction = model.predict(
        pd.DataFrame(
            [features],
            columns=model.feature_names_in_,
        )
    )

    response = {"predictions": prediction.tolist()}
    response_to_mongo(
        response,
    )
    return response

我收到一条错误消息：

TypeError: 'ObjectId' object is not iterable

从我的阅读来看，这似乎是由于 FastAPI 和 Mongo 之间的 BSON/JSON 问题造成的。但是，为什么当我不使用变量时它在第一种情况下起作用？这是由于 FastAPI 的异步特性吗？

Answer 1

根据文档：

插入文档时，会自动出现一个特殊键

"_id"

如果文档尚未包含

"_id"

键，则添加。价值 的

"_id"

在整个集合中必须是唯一的。

insert_one()

返回一个 InsertOneResult 的实例。有关“_id”的更多信息，请参阅 关于 _id 的文档。

因此，在您提供的示例的第二种情况下，当您将字典传递给

insert_one()

函数时，Pymongo 会将从数据库检索数据所需的唯一标识符（即

ObjectId

）添加到您的字典中；因此，当从端点返回响应时，

ObjectId

无法序列化，因为如这个答案中详细描述的，FastAPI默认情况下会使用自动将该返回值转换为与JSON兼容的数据

jsonable_encoder

（确保不可序列化的对象被转换为

str

），然后返回一个

JSONResponse

，使用标准

json

库来序列化数据。

解决方案1

使用此处演示的方法，默认情况下将

ObjectId

转换为

str

，因此，您可以像往常一样在端点内返回

response

。

# place these at the top of your .py file
import pydantic
from bson import ObjectId
pydantic.json.ENCODERS_BY_TYPE[ObjectId]=str

return response # as usual

解决方案2

将加载的

BSON

转储到有效的

JSON

字符串，然后将其重新加载为

dict

，如此处和此处所述。

from bson import json_util
import json

response = json.loads(json_util.dumps(response))
return response

解决方案3

定义自定义

JSONEncoder

，如此处所述，将

ObjectId

转换为

str

：

import json
from bson import ObjectId

class JSONEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, ObjectId):
            return str(o)
        return json.JSONEncoder.default(self, o)


response = JSONEncoder().encode(response)
return response

解决方案4

您可以拥有一个单独的输出模型，无需“ObjectId”(

_id

)字段，如文档中所述。您可以在端点的装饰器中使用参数

response_model

声明用于响应的模型。示例：

from pydantic import BaseModel

class ResponseBody(BaseModel):
    name: str
    age: int


@app.get('/', response_model=ResponseBody)
def main():
    # response sample
    response = {'_id': ObjectId('53ad61aa06998f07cee687c3'), 'name': 'John', 'age': '25'}
    return response

解决方案5

在返回之前从

"_id"

字典中删除

response

条目（请参阅此处了解如何从

dict

中删除密钥）：

response.pop('_id', None)
return response

Answer 2

解决方案 4，来自 Chris 的出色回答，也可以通过函数输出类型提示来完成。因此：

from pydantic import BaseModel


class ResponseBody(BaseModel):
    name: str
    age: int


@app.get('/')
def example() -> ResponseBody:
    # you'd need to await this if you were using Motor (the Async MongoDB Driver)
    return db.my_collection.find_one(...)