FastAPI读取MongoDB用户报错：Input should be a valid list，求优化方案

问题描述 #

我是Python新手，编写的FastAPI接口在读取MongoDB中的所有用户时返回Input should be a valid list错误，触发pydantic_core._pydantic_core.ValidationError，提示projects字段应为有效列表，但数据库中该字段实际为单个字典对象。同时想咨询更优的查询路由实现方式。

相关代码与信息 #

main.py

from typing import List
import json

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from config import collection
from model import Usuario

app = FastAPI()

app.add_middleware( #用于允许连接的参数
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def read_users() -> List[Usuario]:
    itens = [Usuario(**usuario) for usuario in collection.find()]
    print(itens)
    return itens

config.py

from pymongo import MongoClient

client = MongoClient("mongodb://allanvargensDB:vargens@localhost:27017/")

db = client['portfolio']
collection = db['usuario']

model.py

from typing import List

from bson import ObjectId
from pydantic import BaseModel,  Field, BaseConfig

class OID(str):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        try:
            return ObjectId(str(v))
        except Exception:
            raise ValueError("not a valid ObjectId")

class MyBaseModel(BaseModel):
    def __hash__(self):
        return hash((type(self),) + tuple(self.__dict__.values()))

class Projects(MyBaseModel):
    id_project: ObjectId = Field(description='project id', alias='project_id')
    title: str
    imagePath: str
    url: str = ''

    class Config(BaseConfig):
        arbitrary_types_allowed = True
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

class Usuario(BaseModel):
    id_user: ObjectId = Field(description='user id', alias='user_id')
    name: str
    login: str
    password: str
    projects: List[Projects] = Field(alias='projects')

    class Config(BaseConfig):
        arbitrary_types_allowed = True
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

错误信息

pydantic_core._pydantic_core.ValidationError: 1 validation error for Usuario projects
   Input should be a valid list [type=list_type, input_value={'project_id': ObjectId('..., 'url': 'teste.com.br'}, input_type=dict]

数据库文档结构

{
    _id: ObjectId('656e6c2fc0feaabbfb87d63e'),
    name: 'Allan Vargens',
    login: 'allanvargens',
    password: 'password',
    projects: {
        project_id: ObjectId('656e6d7ec0feaabbfb87d640'),
        titulo: 'teste',
        imageUrl: '',
        url: 'teste.com.br'
    },
    user_id: ObjectId('656e6c2fc0feaabbfb87d63e')
}

解决方案 #

一、修复ValidationError问题 #

1. 核心类型不匹配修复

你的Pydantic模型Usuario中projects定义为列表类型，但数据库中该字段是单个字典，这是错误根源。有两种处理方式：

方案A：修正数据库结构（推荐）

将数据库中用户文档的projects字段改为数组格式，符合多项目的业务逻辑：

{
    "_id": ObjectId("656e6c2fc0feaabbfb87d63e"),
    "name": "Allan Vargens",
    "login": "allanvargens",
    "password": "password",
    "projects": [
        {
            "project_id": ObjectId("656e6d7ec0feaabbfb87d640"),
            "titulo": "teste",
            "imageUrl": "",
            "url": "teste.com.br"
        }
    ],
    "user_id": ObjectId("656e6c2fc0feaabbfb87d63e")
}

方案B：修改模型兼容单/多格式

如果暂时无法修改数据库，可给projects字段添加验证器，自动将单个字典转为列表：

from typing import List, Union
from pydantic import field_validator

class Usuario(BaseModel):
    id_user: ObjectId = Field(description='user id', alias='user_id')
    name: str
    login: str
    password: str
    projects: List[Projects] = Field(alias='projects')

    @field_validator('projects', mode='before')
    def ensure_list(cls, v):
        if isinstance(v, dict):
            return [v]
        return v

    class Config(BaseConfig):
        arbitrary_types_allowed = True
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

2. 字段名不匹配修复

模型与数据库的字段名不一致，需添加别名映射：

class Projects(MyBaseModel):
    id_project: ObjectId = Field(description='project id', alias='project_id')
    title: str = Field(alias='titulo')  # 对应数据库的titulo
    imagePath: str = Field(alias='imageUrl')  # 对应数据库的imageUrl
    url: str = ''

    class Config(BaseConfig):
        arbitrary_types_allowed = True
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

二、优化查询路由实现 #

1. 改用异步MongoDB驱动（Motor）

FastAPI是异步框架，同步的pymongo会阻塞事件循环，换成异步驱动motor更高效：

• 安装依赖：pip install motor
• 修改config.py：

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://allanvargensDB:vargens@localhost:27017/")
db = client['portfolio']
collection = db['usuario']

2. 异步路由实现

修改路由为异步逻辑，避免阻塞：

from typing import List
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from config import collection
from model import UsuarioOut

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/", response_model=List[UsuarioOut])
async def read_users():
    users = []
    async for usuario in collection.find():
        try:
            users.append(UsuarioOut.model_validate(usuario))
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"用户数据验证失败: {str(e)}")
    return users

3. 隐藏敏感字段

拆分Pydantic模型，避免返回password等敏感信息：

# 内部模型（包含所有字段）
class UsuarioIn(BaseModel):
    id_user: ObjectId = Field(alias='user_id')
    name: str
    login: str
    password: str
    projects: List[Projects] = Field(alias='projects')

    class Config(BaseConfig):
        arbitrary_types_allowed = True
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

# 对外输出模型（隐藏敏感字段）
class UsuarioOut(BaseModel):
    id_user: str = Field(alias='user_id')
    name: str
    login: str
    projects: List[Projects] = Field(alias='projects')

    class Config(BaseConfig):
        json_encoders = {
            ObjectId: lambda oid: str(oid)
        }

4. 批量数据优化

数据量较大时，用to_list一次性获取所有文档提升效率：

@app.get("/", response_model=List[UsuarioOut])
async def read_users():
    users = await collection.find().to_list(length=None)
    return [UsuarioOut.model_validate(user) for user in users]

内容的提问来源于stack exchange，提问作者learn.boy