如何用SQLAlchemy实现MySQL库存列求和？遇Pydantic验证错误求助

API 探路者

2026-8-3

问题原因分析

你遇到的Pydantic验证错误，核心原因是聚合查询返回的结果（仅单个求和值）和你当前使用的Item模型结构不匹配：Item模型要求id、TotalSupplyQuantity等必填字段，但你的求和SQL只返回sum_stock一个值，自然无法通过模型校验。

解决方案

1. 创建专用的聚合响应模型

新建一个只包含求和结果字段的Pydantic模型，用于接收聚合查询的返回值（可放在schemas.py或直接写在主文件中）：

from pydantic import BaseModel

class StockSumResponse(BaseModel):
    sum_stock: int | None = None

    class Config:
        orm_mode = True

2. 编写SQLAlchemy聚合查询逻辑

在crud.py中实现正确的求和查询，直接返回聚合值：

from sqlalchemy.orm import Session
from sqlalchemy import func
from .models import MwsInventory

def calculate_total_in_stock(db: Session):
    # 使用func.sum执行求和，scalar()直接获取单个结果
    total = db.query(func.sum(MwsInventory.InStockSupplyQuantity)).scalar()
    # 处理空表情况，返回0而非None
    return {"sum_stock": total or 0}

3. 接口层返回专用响应模型

在main.py的接口中，指定返回上述新模型，而非原来的Item模型：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from . import crud, models
from .database import SessionLocal, engine
from .schemas import StockSumResponse

models.Base.metadata.create_all(bind=engine)
app = FastAPI()

# 数据库会话依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/inventory/stock-sum", response_model=StockSumResponse)
def get_stock_sum(db: Session = Depends(get_db)):
    return crud.calculate_total_in_stock(db)