如何使用Flask SQLAlchemy按结果值分组统计各房源评分数量？

有一个包含两张表的数据库，Listing表与Review表为一对多关联，表结构如下：

class Listing(db.Model):
    __tablename__ = 'listings'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(500), index=True)

class Review(db.Model):
    __tablename__ = 'reviews'
    id = db.Column(db.Integer, primary_key=True)
    rating = db.Column(db.Integer())
    listing_id = db.Column(db.Integer, db.ForeignKey('listings.id'))
    listing = db.relationship("Listing", backref="reviews")

目标是统计每个房源1-5星评论的数量，期望结果格式如下：

name, rating, count
Listing A, 5, 10
Listing A, 4, 3
Listing A, 3, 2
Listing B, 5, 6
Listing B, 2, 5

尝试了如下查询，但结果缺少Review.rating字段，无法对应数量所属的评分等级，且未得到预期结构化结果：

listing_summary = db.session \
    .query(Listing.name, db.func.count(Review.review_id)) \
    .outerjoin(Review, Listing.id==Review.listing_id) \
    .order_by(Listing.name.asc()) \
    .group_by(Listing.name) \
    .group_by(Review.rating).all()

for listing in listing_summary:
    print(listing)

当前输出结果：

('Listing A', 59)
('Listing A', 9)
('Listing A', 20)
('Listing A', 31)
('Listing A', 89)
('Listing B', 0)
('Listing C', 0)
('Listing D', 0)
('Listing E', 0)

原本预期按Listing.name和Review.rating分组后得到类似如下的结构化结果：

Listing A, [(5, 59),(4, 9),(3, 20),(2, 31),(1, 89)]

1. 基础修正：获取带评分的统计结果 #

你的查询核心问题是未将Review.rating加入查询字段，且统计时用了不存在的Review.review_id。修正后的基础查询如下：

listing_summary = db.session \
    .query(
        Listing.name,
        Review.rating,
        db.func.count(Review.id).label('count')
    ) \
    .outerjoin(Review, Listing.id == Review.listing_id) \
    .group_by(Listing.name, Review.rating) \
    .order_by(Listing.name.asc(), Review.rating.desc()) \
    .all()

# 按期望格式打印
for item in listing_summary:
    print(f"{item.name}, {item.rating}, {item.count}")

2. 过滤无评论的房源（可选） #

如果不需要显示rating为None且count为0的无评论房源记录，可以改用内连接并过滤空值：

listing_summary = db.session \
    .query(
        Listing.name,
        Review.rating,
        db.func.count(Review.id).label('count')
    ) \
    .join(Review, Listing.id == Review.listing_id)  # 内连接仅保留有评论的房源
    .group_by(Listing.name, Review.rating) \
    .order_by(Listing.name.asc(), Review.rating.desc()) \
    .all()

3. 生成嵌套结构化结果 #

如果想要得到房源名称: [(评分, 数量), ...]的嵌套格式，可以在查询后用Python代码整理结果：

from collections import defaultdict

# 执行查询并过滤空评分
results = db.session \
    .query(Listing.name, Review.rating, db.func.count(Review.id)) \
    .outerjoin(Review, Listing.id == Review.listing_id) \
    .filter(Review.rating.isnot(None)) \
    .group_by(Listing.name, Review.rating) \
    .order_by(Listing.name.asc(), Review.rating.desc()) \
    .all()

# 整理成嵌套结构
structured_result = defaultdict(list)
for name, rating, count in results:
    structured_result[name].append((rating, count))

# 打印嵌套结果
for name, ratings in structured_result.items():
    print(f"{name}, {ratings}")

关键修正点说明 #

• 加入Review.rating字段：让统计结果能对应到具体的评分等级。
• 修正count统计字段：使用Review.id替代不存在的Review.review_id，确保统计准确。
• 简化分组写法：group_by(Listing.name, Review.rating)与连续两次group_by效果一致，代码更简洁。
• 优化排序逻辑：添加Review.rating.desc()让同一房源的评分从高到低排列，结果更直观。

内容的提问来源于stack exchange，提问作者Johnny John Boy