如何在SQLAlchemy多对多关系中避免重复求和

我有三个SQLAlchemy表：TagGroup、Tag和Video。TagGroup与Tag、Tag与Video均通过关联表建立双向多对多关系。我需要构造一个查询，对指定TagGroup.id对应的所有Video的viewCount字段求和。但如果同一个视频关联了同一TagGroup下的多个不同Tag，Video.viewCount会被重复求和，希望找到避免该问题的方法。

示例结构：

Video1
                /
           Tag1
         /      \
 TagGroup         Video2 (会被重复求和)
         \      /
           Tag2
                \
                  Video3

注：为简化内容，已移除所有无关字段。

TagGroup 模型 #

class TagGroup(Base):
    __tablename__ = "tag_groups"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )

    tags = relationship(
        "Tag",
        secondary=tags_and_groups_association_table,
        back_populates="groups",
    )

TagGroup - Tag 关联表 #

tags_and_groups_association_table = Table(
    "tags_and_groups_association_table",
    Base.metadata,
    Column("tags_id", ForeignKey("tags.id"), primary_key=True),
    Column("tag_groups_id", ForeignKey("tag_groups.id"), primary_key=True),
    PrimaryKeyConstraint('tags_id', 'tag_groups_id')  # 避免重复
)

Tag 模型 #

class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )

    in_videos = relationship("Video", secondary=video_tags, back_populates="tags",lazy="select")

    groups = relationship(
        "TagGroup",
        secondary=tags_and_groups_association_table,
        back_populates="tags",
    )

Tag - Video 关联表 #

video_tags = Table(
    'video_tags', Base.metadata,
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True),
    Column('video_id', Integer, ForeignKey('videos.id'), primary_key=True)
)

Video 模型 #

class Video(Base):
    __tablename__ = "videos"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )
    viewCount: Mapped[int] = mapped_column(
        BigInteger,
        nullable=False,
        default=0,
    )

    tags = relationship("Tag", secondary=video_tags, back_populates="in_videos", lazy="select")

尝试过的无效查询 #

select(
    TagGroup.id,
    func.SUM(Video.viewCount),
    func.COUNT(distinct(Video.id)),  # 按唯一ID去重的计数是正常的
).select_from(
    TagGroup.id
).where(
    TagGroup.id.in_(ids) # 指定TagGroup ID列表
).join(
    tags_and_groups_association_table,
    TagGroup.id == tags_and_groups_association_table.c.tag_groups_id
).join(
    Tag,
    tags_and_groups_association_table.c.tags_id == Tag.id
).join(
    video_tags_model,
    Tag.id == video_tags_model.c.tag_id,
).join(
    Video,
    video_tags_model.c.video_id == Video.id
).group_by(
    TagGroup.id,
)

解决方案 #

问题根源是多对多关联产生的笛卡尔积，同一视频会因关联多个Tag被多次统计。核心是先确保每个视频在对应TagGroup下仅出现一次，再求和。

方法一：子查询先获取唯一视频集合 #

先通过子查询筛选出每个TagGroup对应的所有唯一视频，再基于这些视频求和：

# 子查询：获取每个TagGroup关联的唯一视频
unique_videos_subq = select(
    TagGroup.id.label("tag_group_id"),
    Video.id.label("video_id"),
    Video.viewCount
).select_from(TagGroup)
.join(tags_and_groups_association_table)
.join(Tag)
.join(video_tags)
.join(Video)
.where(TagGroup.id.in_(ids))
.distinct(TagGroup.id, Video.id)  # 保证同一分组下每个视频只出现一次

# 主查询：对唯一视频的viewCount求和
final_query = select(
    unique_videos_subq.c.tag_group_id,
    func.SUM(unique_videos_subq.c.viewCount).label("total_views"),
    func.COUNT(unique_videos_subq.c.video_id).label("video_count")
).select_from(unique_videos_subq)
.group_by(unique_videos_subq.c.tag_group_id)

方法二：双层分组聚合 #

先按TagGroup和Video分组（确保每个视频只算一次），再对TagGroup分组求和：

final_query = select(
    TagGroup.id,
    func.SUM(Video.viewCount).label("total_views"),
    func.COUNT(Video.id).label("video_count")
).select_from(TagGroup)
.join(tags_and_groups_association_table)
.join(Tag)
.join(video_tags)
.join(Video)
.where(TagGroup.id.in_(ids))
.group_by(TagGroup.id, Video.id)  # 先按视频去重
.with_entities(
    TagGroup.id,
    func.sum(Video.viewCount).label("total_views"),
    func.count(Video.id).label("video_count")
)
.group_by(TagGroup.id)  # 再按分组求和

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