如何从SQLAlchemy多对多关联中获取指定格式的事物标签列表

我有如下带多对多关联关系的SQLAlchemy数据结构：

things_tags = Table("things_to_tags", Base.metadata,
                    Column('id', Integer, primary_key=True),
                    Column('thing_id', Integer, ForeignKey('things_table.id')),
                    Column('tag_id', Integer, ForeignKey('tags_table.id')),
                   )

class Thing(Base):
    __tablename__ = 'things_table'
    id = Column(Integer, primary_key=True)    
    title = Column(String)    
    tags = relationship("Tag", secondary = things_tags, back_populates='things')

class Tag(Base):
    __tablename__ = 'tags_table'    
    id = Column(Integer, primary_key=True)
    name = Column(String)    
    things = relationship("Thing", secondary = things_tags)

我希望从数据库中检索所有Thing，并以如下格式返回其关联的Tag：

[
 (thing1, [tag2, tag5, tag6]),
 (thing2, [tag1, tag3]),
 ...
]

我尝试用以下语句查询数据库：

session.execute(select(Thing.title,Tag.name).join(Thing.tags)).all()

但得到的结果格式如下：

[(thing1, tag2),
 (thing1, tag5),
 (thing1, tag6),
 (thing2, tag1),
 (thing2, tag3),
...
]

我用以下代码几乎得到了想要的结果：

for x in session.query( Thing ).all():
    print(x.title, x.tags)

但这不是正确做法：一是query()已被SQLAlchemy 2.0弃用，二是每次调用x.tags都会触发额外查询（N+1问题），大数据量下会产生大量数据库请求，性能极差。

方法一：预加载关联（推荐，适配SQLAlchemy 2.0+） #

使用SQLAlchemy 2.0的select配合selectinload预加载Thing的tags关联，一次性加载所有数据，彻底避免N+1查询：

from sqlalchemy.orm import selectinload

# 预加载所有Thing及其关联的tags
things = session.execute(
    select(Thing).options(selectinload(Thing.tags))
).scalars().all()

# 整理成目标格式
result = [
    (thing.title, [tag.name for tag in thing.tags])
    for thing in things
]

• selectinload会执行两次高效查询：一次获取所有Thing，一次获取这些Thing对应的所有Tag，随后在内存中完成关联，性能远优于N+1模式。
• 若偏好单次JOIN查询，可使用joinedload，但需配合distinct()避免重复的Thing实例：

from sqlalchemy.orm import joinedload

things = session.execute(
    select(Thing).options(joinedload(Thing.tags)).distinct()
).scalars().all()

result = [
    (thing.title, [tag.name for tag in thing.tags])
    for thing in things
]

方法二：数据库层面分组聚合（直接返回目标结构） #

如果希望直接从数据库获取聚合后的结果，可使用对应数据库的聚合函数配合分组查询，以PostgreSQL为例：

from sqlalchemy import func, select

result = session.execute(
    select(
        Thing.title,
        func.array_agg(Tag.name).label('tags')
    )
    .join(Thing.tags)
    .group_by(Thing.id, Thing.title)
).all()

这种方式让数据库直接完成分组聚合，返回的结果就是(title, [tag1, tag2,...])的目标格式，无需内存二次处理。不同数据库的聚合函数有所差异：

• MySQL/SQLite 使用func.group_concat，返回的是逗号分隔的字符串，需自行拆分转换为列表。

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