基于MongoID初始化Pydantic子模型关联的实现方案（Pydantic+FastAPI+MongoDB）

1. 定义适配MongoDB的Pydantic模型 #

首先处理MongoDB的ObjectId类型，通过Pydantic的PyObjectId实现类型兼容，再分别定义用户模型和包含嵌套用户的项目模型：

from pydantic import BaseModel, Field
from pydantic_core import PyObjectId
from typing import Optional

# 用户模型
class User(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id")
    name: str
    email: str
    user_level: int

    class Config:
        populate_by_name = True
        json_encoders = {PyObjectId: str}  # 将ObjectId序列化为字符串，避免JSON序列化报错

# 包含嵌套用户的项目模型
class MainProjectData(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id")
    name: str
    customer: str
    user: Optional[User]  # 用嵌套User模型替代原有的user_id字段

    class Config:
        populate_by_name = True
        json_encoders = {PyObjectId: str}

2. 修改MongoDB查询逻辑，关联用户集合 #

使用MongoDB的$lookup聚合操作，将mainProjects集合与users集合通过user_id关联，把匹配的用户数据嵌入项目结果中：

from pymongo import MongoClient

# 初始化MongoDB客户端
client = MongoClient("mongodb://localhost:27017/")
db = client["your_database_name"]  # 替换为你的实际数据库名称

def get_projects_with_users():
    # 聚合查询管道：关联用户集合并展开结果
    pipeline = [
        {
            "$lookup": {
                "from": "users",          # 要关联的用户集合名
                "localField": "user_id",  # 项目集合中的关联字段
                "foreignField": "_id",    # 用户集合中的关联字段
                "as": "user"              # 结果中存储用户数据的字段名
            }
        },
        {
            "$unwind": {
                "path": "$user",
                "preserveNullAndEmptyArrays": True  # 保留无对应用户的项目
            }
        }
    ]
    # 执行聚合查询并返回结果列表
    return list(db["mainProjects"].aggregate(pipeline))

3. 更新Flask API视图函数 #

将关联查询得到的数据传入Pydantic模型验证，再序列化为JSON返回：

from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/projects", methods=["GET"])
def get_projects():
    projects_raw = get_projects_with_users()
    # 用Pydantic模型解析原始数据
    projects = [MainProjectData(**item) for item in projects_raw]
    # 转换为JSON格式返回，保留MongoDB字段别名
    return jsonify([project.model_dump(by_alias=True) for project in projects])

if __name__ == "__main__":
    app.run(debug=True)

关键注意事项 #

• ObjectId序列化：必须在Pydantic模型的Config中配置json_encoders，否则无法将MongoDB的ObjectId类型转换为可JSON序列化的字符串。
• 关联字段匹配：确保$lookup中的localField和foreignField对应正确，user_id是项目集合中的ObjectId字段，_id是用户集合的主键。
• 空用户场景处理：$unwind的preserveNullAndEmptyArrays参数避免因用户不存在导致对应项目被过滤。
• 字段别名映射：Pydantic模型中用alias="_id"对应MongoDB的_id字段，同时开启populate_by_name=True确保字段映射正确。

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