FastAPI同时使用两个Pydantic响应模型报错问题
FastAPI接口多表查询响应模型报错问题
我用FastAPI实现接口,初始版本的main.py中,/licence_ol/接口使用list[EvergreenOutput]作为响应模型可正常运行。当修改响应模型为list[EvergreenOutput, MappingOutput],并调整查询语句同时查询Evergreen和Mapping表后,接口返回500内部服务器错误,报错信息显示EvergreenOutput的product_id字段缺失。去掉类型注解时,响应结构为包含Evergreen和Mapping对象的列表。
初始可运行的main.py代码
@app.get( "/licence_ol/", summary="Query Licence info by product id", response_description="Successful Query", response_model=list[EvergreenOutput], tags=[Tags.items] ) def get_product_id(session: Session = Depends(get_session))-> list[EvergreenOutput]: query = select(Evergreen).where(Evergreen.product_id == Mapping.eim_product_id).limit(200) return session.exec(query).fetchall()
schemas.py代码
from typing import Optional from datetime import date from sqlmodel import SQLModel, Field class Mapping(SQLModel,table=True): id_seq: Optional[int] = Field(default=None,primary_key=True) eim_product_id: Optional[int] = None mem_product_id: Optional[int] = None vendor: Optional[str] = None name: Optional[str] = None eim_name: Optional[str] = None product_alias: Optional[str] = None lev: Optional[int] = None manufacturer: Optional[str] = None class MappingOutput(SQLModel): eim_product_id: Optional[int] = None mem_product_id: Optional[int] = None vendor: Optional[str] = None name: Optional[str] = None eim_name: Optional[str] = None product_alias: Optional[str] = None lev: Optional[int] = None manufacturer: Optional[str] = None class Evergreen(SQLModel,table=True): id_seq: Optional[int] = Field(default=None,primary_key=True) product_id: int phase_type: Optional[str] = None phase_start: Optional[date] = None phase_end: Optional[date] = None product_name: Optional[str] = None software_product_version_name: Optional[str] = None software_product_version_id: Optional[int] = None class EvergreenOutput(SQLModel): product_id: int phase_type: Optional[str] = None phase_start: Optional[date] = None phase_end: Optional[date] = None product_name: Optional[str] = None software_product_version_name: Optional[str] = None software_product_version_id: Optional[int] = None
修改后报错的main.py代码
@app.get( "/licence_ol/", summary="Query Licence info by product id", response_description="Successful Query", response_model=list[EvergreenOutput,MappingOutput], tags=[Tags.items] ) def get_product_id(session: Session = Depends(get_session))-> list[EvergreenOutput,MappingOutput]: query = select(Evergreen,Mapping).where(Evergreen.product_id == Mapping.eim_product_id).limit(200) return session.exec(query).fetchall()
错误信息
INFO: 127.0.0.1:53172 - "GET /licence_ol/ HTTP/1.1" 500 Internal Server Error ERROR: Exception in ASGI application Traceback (most recent call last): File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 404, in run_asgi result = await app( # type: ignore[func-returns-value] File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 78, in __call__ return await self.app(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\fastapi\applications.py", line 270, in __call__ await super().__call__(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\applications.py", line 124, in __call__ await self.middleware_stack(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\middleware\errors.py", line 184, in __call__ raise exc File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\middleware\errors.py", line 162, in __call__ await self.app(scope, receive, _send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\middleware\exceptions.py", line 75, in __call__ raise exc File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\middleware\exceptions.py", line 64, in __call__ await self.app(scope, receive, sender) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\fastapi\middleware\asyncexitstack.py", line 21, in __call__ raise e File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\fastapi\middleware\asyncexitstack.py", line 18, in __call__ await self.app(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\routing.py", line 680, in __call__ await route.handle(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\routing.py", line 275, in handle await self.app(scope, receive, send) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\starlette\routing.py", line 65, in app response = await func(request) File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\fastapi\routing.py", line 253, in app content = await serialize_response( File "C:\Users\45291029\Documents\venv\ml\lib\site-packages\fastapi\routing.py", line 139, in serialize_response raise ValidationError(errors, field.type_) pydantic.error_wrappers.ValidationError: 200 validation errors for EvergreenOutput response -> 0 -> product_id field required (type=value_error.missing)
无类型注解时的响应结构
[ { "Evergreen": { "id_seq": 344, "phase_start": "2011-07-22", "product_name": "HTTP Server (IHS)", "software_product_version_id": 359483, "phase_end": "2018-04-30", "product_id": 359496, "phase_type": "Generally Available", "software_product_version_name": "IBM HTTP Server v8.0" }, "Mapping": { "eim_product_id": 359496, "mem_product_id": 19, "name": "HTTP Server (IHS)", "product_alias": "IHS", "manufacturer": "IBM", "id_seq": 0, "vendor": "IBM", "eim_name": "HTTP Server (IHS)", "lev": 100 } } ]
解决方案
问题根源
- 响应模型写法错误:
list[EvergreenOutput, MappingOutput]不是合法的类型注解,无法表示“列表元素同时包含两种模型”的结构。FastAPI会尝试把每个返回元素适配成EvergreenOutput,但实际返回的是包含Evergreen和Mapping的组合对象,导致字段匹配失败。 - 查询返回结构不匹配:
select(Evergreen, Mapping)返回的是包含两个模型实例的组合对象(或元组),和定义的单个输出模型结构完全不符。
修复步骤
- 定义组合响应模型:在
schemas.py中新增一个包含两个输出模型的组合类:
class CombinedLicenceOutput(SQLModel): Evergreen: EvergreenOutput Mapping: MappingOutput
- 修改接口的响应模型和返回类型:更新
main.py中的接口定义,使用组合模型作为列表元素:
from schemas import CombinedLicenceOutput @app.get( "/licence_ol/", summary="Query Licence info by product id", response_description="Successful Query", response_model=list[CombinedLicenceOutput], tags=[Tags.items] ) def get_product_id(session: Session = Depends(get_session))-> list[CombinedLicenceOutput]: query = select(Evergreen, Mapping).where(Evergreen.product_id == Mapping.eim_product_id).limit(200) # 将查询结果转换为符合CombinedLicenceOutput的结构 results = [] for evergreen, mapping in session.exec(query).fetchall(): results.append({ "Evergreen": evergreen, "Mapping": mapping }) return results
部分版本的SQLModel会自动将查询返回的元组转换为带键的字典,此时可以直接返回session.exec(query).fetchall(),但显式转换结构更稳妥。
修改后,响应模型与实际返回结构完全匹配,可通过Pydantic验证,接口即可正常返回数据。
内容的提问来源于stack exchange,提问作者moth
相关产品推荐
相关产品推荐

