FastAPI写入MongoDB成功却触发异常返回问题求助

问题根源 #

1. 错误使用await调用同步函数：add_building是普通同步函数，并非异步协程，但接口中用await调用它，会触发TypeError: object InsertOneResult can't be used in 'await' expression，导致进入异常分支。此时数据已经完成插入（同步函数已执行完insert_one），但后续的await操作抛出了异常。
2. 返回值无法序列化：即使去掉await，add_building返回的InsertOneResult对象无法被FastAPI自动序列化为JSON格式，返回时会引发序列化异常，同样进入except分支。

解决方案 #

方案1：修正同步函数调用方式（线程池包装）

由于PyMongo是同步客户端，在FastAPI异步接口中调用同步阻塞函数需要用run_in_threadpool包装，避免阻塞事件循环，同时修正await的错误使用：

1. 导入线程池工具：

from fastapi.concurrency import run_in_threadpool

2. 修改接口代码：

@router.post("/add_building")
async def post_building(building: BuildingPayloadSchema):
    try:
        # 用线程池包装同步函数调用
        result = await run_in_threadpool(mongodb_connector.add_building, building)
        # 返回可序列化的结果（提取inserted_id）
        return {"status": "success", "inserted_id": str(result.inserted_id)}
    except Exception as e:
        # 捕获具体异常便于调试
        return {"status": "error", "message": f"Something went wrong: {str(e)}", "code": HTTPStatus.UNPROCESSABLE_ENTITY}

3. 简化add_building函数（去掉多余的JSON序列化）：

def add_building(self, building_data):
    # 直接用Pydantic模型的dict()方法转换为字典
    building_dict = building_data.dict()
    insert_result = self._collection.insert_one(building_dict)
    return insert_result

方案2：改用异步MongoDB客户端（Motor）

如果想彻底使用异步操作，替换PyMongo为Motor（MongoDB官方异步Python驱动）：

1. 安装Motor：

pip install motor

2. 修改MongoDB连接类为异步实现：

from motor.motor_asyncio import AsyncIOMotorClient

class MongoDBConnector:
    def __init__(self, mongo_url, db_name, collection_name):
        self._mongo_url = mongo_url
        self._db_name = db_name
        self._collection_name = collection_name
        self._client = None
        self._db = None
        self._collection = None

    async def connect(self):
        self._client = AsyncIOMotorClient(self._mongo_url)
        print("MongoDB connected...")
        self._db = self._client[self._db_name]
        print(f"{self._db_name} database has been created...")
        self._collection = self._db[self._collection_name]
        print(f"Building collection has been created...")

    async def add_building(self, building_data):
        building_dict = building_data.dict()
        insert_result = await self._collection.insert_one(building_dict)
        return insert_result

3. 修改接口代码（此时add_building是异步函数，可直接用await）：

@router.post("/add_building")
async def post_building(building: BuildingPayloadSchema):
    try:
        result = await mongodb_connector.add_building(building)
        return {"status": "success", "inserted_id": str(result.inserted_id)}
    except Exception as e:
        return {"status": "error", "message": f"Something went wrong: {str(e)}", "code": HTTPStatus.UNPROCESSABLE_ENTITY}

额外建议 #

• 不要使用空的except:，捕获具体异常并打印错误信息，方便定位问题。
• Pydantic模型直接用dict()方法转换为字典即可，无需先json.dumps再json.loads，避免不必要的性能消耗。

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