如何将FastAPI的Pydantic模型关联至已有MongoDB集合？

1. 修正Schema转换函数 #

原转换函数中_id的处理有误，MongoDB的_id为ObjectId类型，直接用str()转换即可，调整后确保输出字段与Pydantic模型匹配：

def stock_entity(db_item) -> dict:
    return {
        "id": str(db_item["_id"]),
        "cotacao": db_item["cotacao"],
        "pl": db_item["pl"],
        "pvp": db_item["pvp"],
        "psr": db_item["psr"],
        "dy": db_item["dy"],
    }

def stock_list_entity(db_items) -> list:
    return [stock_entity(item) for item in db_items]

2. 调整Pydantic模型 #

MongoDB文档自带_id字段，需在模型中添加可选的id字段（创建文档时无需传入），同时开启orm_mode支持从字典/ORM对象读取数据：

from pydantic import BaseModel
from typing import Optional

class Stock(BaseModel):
    id: Optional[str] = None
    cotacao: int
    pl: int
    pvp: int
    psr: int
    dy: int

    class Config:
        orm_mode = True

3. 实现单条文档的GET路由 #

添加通过id查询单条文档的接口，验证ID格式合法性并处理文档不存在的情况，结合转换函数返回符合模型的结构：

from fastapi import APIRouter, HTTPException
from config.db import connection
from models.stock import Stock
from schemas.stock import stock_entity, stock_list_entity
from bson.objectid import ObjectId

stock_router = APIRouter(
    tags=['stock'],
    prefix='/stock'
)

# 获取所有文档
@stock_router.get('/')
async def get_stocks():
    stocks = connection.local.stock.find()
    return stock_list_entity(stocks)

# 获取单条文档
@stock_router.get('/{stock_id}')
async def get_single_stock(stock_id: str):
    if not ObjectId.is_valid(stock_id):
        raise HTTPException(status_code=400, detail="无效的股票ID格式")
    
    stock = connection.local.stock.find_one({"_id": ObjectId(stock_id)})
    if not stock:
        raise HTTPException(status_code=404, detail="未找到对应股票")
    
    return stock_entity(stock)

关联逻辑说明 #

• Schema转换函数作为中间层，将MongoDB原始文档（含_id等特定字段）转换成符合Pydantic模型的字典结构
• Pydantic模型定义数据结构与类型约束，确保API返回数据的规范性
• 路由中通过MongoDB查询方法获取数据后，经转换函数处理再返回，完成集合与模型的绑定

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