FastAPI中如何将点赞统计及用户点赞查询合并为主查询？

1. 先明确模型结构（示例） #

假设你的核心模型定义大致如下：

from sqlalchemy import Column, Integer, String, ForeignKey, Boolean
from sqlalchemy.orm import relationship, Session
from sqlalchemy.sql import func

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)

class Prize(Base):
    __tablename__ = "prizes"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    review_id = Column(Integer, ForeignKey("reviews.id"))
    review = relationship("Review", back_populates="prize")

class Review(Base):
    __tablename__ = "reviews"
    id = Column(Integer, primary_key=True)
    content = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User")
    prize = relationship("Prize", uselist=False)

class Like(Base):
    __tablename__ = "likes"
    id = Column(Integer, primary_key=True)
    review_id = Column(Integer, ForeignKey("reviews.id"))
    user_id = Column(Integer, ForeignKey("users.id"))
    is_liked = Column(Boolean)  # True=点赞，False=点踩

2. 定义子查询 #

点赞统计子查询 #

统计每个Review的总点赞、点踩数量：

likes_query = (
    session.query(
        Like.review_id,
        func.sum(func.cast(Like.is_liked, Integer)).label("total_likes"),
        func.sum(func.cast(~Like.is_liked, Integer)).label("total_dislikes")
    )
    .group_by(Like.review_id)
    .subquery()
)

用户点赞状态子查询 #

查询当前用户对每个Review的互动状态（无互动则返回None）：

current_user_id = 1  # 替换为实际当前用户ID

user_like_query = (
    session.query(
        Like.review_id,
        Like.is_liked.label("user_liked_status")
    )
    .filter(Like.user_id == current_user_id)
    .subquery()
)

3. 整合到主查询 #

通过outerjoin关联子查询，确保没有点赞数据的Review也能被正常返回：

main_query = (
    session.query(
        Review,
        User,
        Prize,
        likes_query.c.total_likes,
        likes_query.c.total_dislikes,
        user_like_query.c.user_liked_status
    )
    .join(User, Review.author_id == User.id)
    .outerjoin(Prize, Review.id == Prize.review_id)
    .outerjoin(likes_query, Review.id == likes_query.c.review_id)
    .outerjoin(user_like_query, Review.id == user_like_query.c.review_id)
)

4. 获取并处理结果 #

执行查询后，结果是包含Review、User、Prize和子查询字段的元组，直接解构即可使用：

results = main_query.all()

for review, author, prize, total_likes, total_dislikes, user_liked in results:
    # 转换为接口返回所需的结构
    review_data = {
        "id": review.id,
        "content": review.content,
        "author": {"id": author.id, "name": author.name},
        "prize": {"id": prize.id, "name": prize.name} if prize else None,
        "total_likes": total_likes or 0,  # 无数据时默认0
        "total_dislikes": total_dislikes or 0,
        "user_liked": user_liked  # None表示用户未互动
    }
    print(review_data)

额外说明 #

• 若使用SQLAlchemy 2.0+，可改用select语法，逻辑一致：

from sqlalchemy import select

likes_subq = select(
    Like.review_id,
    func.sum(func.cast(Like.is_liked, Integer)).label("total_likes"),
    func.sum(func.cast(~Like.is_liked, Integer)).label("total_dislikes")
).group_by(Like.review_id).subquery()

main_stmt = select(
    Review, User, Prize,
    likes_subq.c.total_likes, likes_subq.c.total_dislikes,
    user_like_subq.c.user_liked_status
).join(User, Review.author_id == User.id)\
 .outerjoin(Prize)\
 .outerjoin(likes_subq, Review.id == likes_subq.c.review_id)\
 .outerjoin(user_like_subq, Review.id == user_like_subq.c.review_id)

results = session.execute(main_stmt).all()

• 必须用outerjoin而非join，否则会过滤掉没有点赞数据或用户未互动的Review。

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