如何在SQLAlchemy查询中新增count结果列并顺利通过Pydantic校验

解决方案 #

直接修改CRUD逻辑即可同时解决两个问题，修改后的代码如下：

def get_stories(db: Session):
    results = db.query(models.Stories, func.count(models.Prizes.id).label("prizes_count"))\
        .outerjoin(models.Prizes)\
        .group_by(models.Stories.id)\
        .all()
    # 为Stories实例动态绑定计数字段
    for story, count in results:
        story.prizes_count = count
    return [item[0] for item in results]

修改说明 #

• 解决prizes_count=0的记录丢失问题：
  • 将原内连接join替换为左外连接outerjoin，保留所有故事记录，无关联奖品的记录不会被过滤
  • 计数逻辑改为func.count(models.Prizes.id)，左外连接下无关联奖品的记录对应Prizes.id为NULL，count(NULL)会返回0，符合计数需求
• 解决Pydantic校验错误：
  • 遍历查询返回的元组，将计数值赋值给Stories实例的动态属性prizes_count，Pydantic的orm_mode支持直接读取实例的动态属性，无需修改原模型定义
  • 最终仅返回Stories实例列表，完全匹配StoryFullInfo的序列化规则

可选优化方案（全局复用计数） #

如果多处查询都需要用到prizes_count字段，可以直接在Stories模型中定义列属性，无需每次手动写关联查询：

from sqlalchemy import select, func
class Stories(Base):
    # 原有字段和关系保持不变
    # 新增计数字段属性
    prizes_count = column_property(
        select(func.count(Prizes.id))
        .where(Prizes.story_id == id)
        .scalar_subquery()
    )

修改后CRUD逻辑可以简化为：

def get_stories(db: Session):
    return db.query(models.Stories).all()

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