如何用SQLAlchemy、Pydantic和FastAPI处理聚合查询结果——API返回部门聚合值时的Pydantic映射问题排查

我来帮你搞定这个Pydantic验证错误的问题，看了你的代码和报错信息，根源在于两个关键点：聚合查询的字段命名不匹配，以及Decimal类型到int的类型转换。

问题分析 #

1. 字段名不匹配：你的SQLAlchemy查询中使用了func.sum(tbl.count)这类聚合函数，但没有给这些聚合结果指定别名。SQLAlchemy会自动给这些聚合列生成默认名称（比如sum_1、sum_2），而你的Pydantic模型Overview里定义的字段是count、item1等，导致Pydantic找不到对应字段，所以报field required错误。
2. 类型不匹配：SQLAlchemy的sum函数返回的是Decimal类型（从你打印的查询结果也能看到Decimal('25')这样的值），但你的Pydantic模型里这些字段定义的是int类型，所以类型验证失败，报value is not a valid integer。

解决方案 #

步骤1：给聚合函数添加别名 #

修改crud.py中的查询逻辑，给每个func.sum()添加.label()，让返回的字段名和Pydantic模型完全一致：

from sqlalchemy.orm import Session
from sqlalchemy import func, Integer  # 新增Integer导入用于类型转换
from . import models, schemas
import datetime

def get_overview(db: Session, client_id: str):
    text = '1000001,1000002,2000001'
    depts = text.split(',')
    tbl = models.Overview
    overview = db.query(
        tbl.client_id,
        tbl.deptCode,
        tbl.deptName,
        func.sum(tbl.count).label("count"),
        func.sum(tbl.item1).label("item1"),
        func.sum(tbl.item2).label("item2"),
        func.sum(tbl.item3).label("item3"),
        func.sum(tbl.item4).label("item4"),
        func.sum(tbl.item5).label("item5"),
        func.sum(tbl.item6).label("item6")
    ).filter(tbl.client_id == client_id).filter(tbl.deptCode.in_(depts)).group_by(tbl.client_id, tbl.deptCode, tbl.deptName).all()
    return overview

步骤2：处理Decimal到int的类型转换 #

有两种方式可以解决这个类型问题，选其中一种即可：

方式A：在查询时直接转换为int

在func.sum()外层包裹func.cast()，把聚合结果直接转成Integer类型：

# 替换crud.py中的聚合字段部分
func.cast(func.sum(tbl.count), Integer).label("count"),
func.cast(func.sum(tbl.item1), Integer).label("item1"),
func.cast(func.sum(tbl.item2), Integer).label("item2"),
func.cast(func.sum(tbl.item3), Integer).label("item3"),
func.cast(func.sum(tbl.item4), Integer).label("item4"),
func.cast(func.sum(tbl.item5), Integer).label("item5"),
func.cast(func.sum(tbl.item6), Integer).label("item6")

方式B：在Pydantic模型中添加类型转换验证器

修改schemas.py，给需要转换的字段添加validator，自动把Decimal转成int：

from pydantic import BaseModel, validator
from decimal import Decimal
import datetime

class Overview(BaseModel):
    client_id: str
    deptCode: str
    deptName: str
    count: int
    item1: int
    item2: int
    item3: int
    item4: int
    item5: int
    item6: int

    @validator('count', 'item1', 'item2', 'item3', 'item4', 'item5', 'item6')
    def decimal_to_int(cls, value):
        if isinstance(value, Decimal):
            return int(value)
        return value

    class Config:
        orm_mode = True

验证修改效果 #

做完上面的修改后，你的查询结果会：

1. 拥有和Pydantic模型完全匹配的字段名，解决field required错误；
2. 聚合值的类型会变成int（或者被Pydantic自动转成int），解决类型验证失败的问题。

这样你的API就能正常返回符合要求的响应了。

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