如何使用 SQLAlchemy、pydantic 和 FastAPI 处理聚合查询结果

Question

我想创建一个 API returns 每个部门的聚合值作为响应。我是 FastAPI 和 pydantic 的新手，希望得到任何支持。在 pydantic 中映射查询结果的部分似乎工作不正常。

main.py

中的获取方法

@app.get("/api/{client_id}", response_model=List[schemas.Overview])
def read_overview(client_id: str, db: Session = Depends(get_db)):
    db_overview = crud.get_overview(db, client_id=client_id)
    if db_overview is None:
        raise HTTPException(status_code=404, detail="Overview not found")
    return db_overview

crud.py 获取1000001、1000002、2000001三个部门代码对应的数据，按client_id和deptCode

聚合

from sqlalchemy.orm import Session
from sqlalchemy import func
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), func.sum(tbl.item1), func.sum(tbl.item2), func.sum(tbl.item3), func.sum(tbl.item4), func.sum(tbl.item5), func.sum(tbl.item6)).\
        filter(tbl.client_id==client_id). \
        filter(tbl.deptCode.in_(depts)). \
        group_by(tbl.client_id, tbl.deptCode, tbl.deptName).all()
    # print(“overview”, overview)
    return overview

如果我用 print() 检查 QUERY 检索的结果，我可以看到按预期检索了三行。

结果 - 使用 print() 在终端上显示概览

[('C012345', '1000001', 'A地域', Decimal('25'), Decimal('7'), Decimal('0'), Decimal('2'), Decimal('0'), Decimal('0'), Decimal('0')), ('C012345', '1000002', 'Z地域', Decimal('55'), Decimal('15'), Decimal('2'), Decimal('12'), Decimal('0'), Decimal('0'), Decimal('0')), ('C012345', '2000001', 'あ小学校', Decimal('15'), Decimal('5'), Decimal('0'), Decimal('2'), Decimal('0'), Decimal('0'), Decimal('0'))]

但是，我收到错误消息“value is not a valid integer”和“field required”，导致错误如下。错误

  File "/Users/junya/mr2edu_server/venv/lib/python3.9/site-packages/fastapi/routing.py", line 137, in serialize_response
    raise ValidationError(errors, field.type_)
pydantic.error_wrappers.ValidationError: 21 validation errors for Overview
response -> 0 -> count
  value is not a valid integer (type=type_error.integer)
response -> 0 -> item1
  field required (type=value_error.missing)
response -> 0 -> item2
  field required (type=value_error.missing)
response -> 0 -> item3
  field required (type=value_error.missing)
response -> 0 -> item4
  field required (type=value_error.missing)
response -> 0 -> item5
  field required (type=value_error.missing)
response -> 0 -> item6
  field required (type=value_error.missing)

…. continued ...

我也显示schemas.py和models.py。

schemas.py

from pydantic import BaseModel
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

    class Config:
        orm_mode = True

models.py

from sqlalchemy import Column, Date, Integer, String
from sqlalchemy.orm import relationship

from .database import Base

class Overview(Base):
    __tablename__ = "overview"
    id = Column(String, primary_key=True, index=True)
    client_id = Column(String)
    date = Column(Date)
    deptCode = Column(String)
    deptName = Column(String)
    level = Column(Integer)
    count = Column(Integer)
    item1 = Column(Integer)
    item2 = Column(Integer)
    item3 = Column(Integer)
    item4 = Column(Integer)
    item5 = Column(Integer)
    item6 = Column(Integer)

Answer 1

将 label() 添加到与您希望将结果分配到的字段名称相匹配的每个聚合函数调用中：

db.query(
    tbl.client_id,
    tbl.deptCode,
    tbl.deptName,
    func.sum(tbl.count).label("count"), 
    func.sum(tbl.item1).label("item1"),
)

多亏了这些标签，每个结果 sqlalchemy.engine.row.Row 由 FastAPI 使用 fastapi.encoders.jsonable_encoder 在内部转换为字典将包含所需的值。

如何使用 SQLAlchemy、pydantic 和 FastAPI 处理聚合查询结果

How to handle aggregated query results with SQLAlchemy, pydantic and FastAPI

python

sqlalchemy

mysql-python

pydantic

fastapi