SQLAlchemy多对多关联表查询：如何获取模块名称及关联组

问题场景 #

现有Module、ModuleGroup两张表，通过中间表ModuleGroupMember建立多对多关系，模型定义如下：

Module模型 #

from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import mapped_column, relationship

from root.db.ModelBase import ModelBase


class Module(ModelBase):
    __tablename__ = 'module'
    pk = mapped_column(Integer, primary_key=True)
    description = mapped_column(String)
    is_active = mapped_column(Boolean)
    name = mapped_column(String, unique=True, nullable=False)
    ngen_cal_active = mapped_column(String)
    groups = relationship("ModuleGroup", secondary="module_group_member", back_populates="modules", lazy="joined")

ModuleGroup模型 #

from sqlalchemy import String, Integer, Boolean
from sqlalchemy.orm import mapped_column, relationship

from root.db.ModelBase import ModelBase


class ModuleGroup(ModelBase):
    __tablename__ = 'module_group'
    pk = mapped_column(Integer, primary_key=True)
    description = mapped_column(String)
    is_active = mapped_column(Boolean)
    name = mapped_column(String, unique=True, nullable=False)
    modules = relationship("Module", secondary="module_group_member", back_populates="groups", lazy="joined")

ModuleGroupMember模型 #

from sqlalchemy import String, Integer, Boolean, ForeignKey
from sqlalchemy.orm import mapped_column

from root.db.ModelBase import ModelBase


class ModuleGroupMember(ModelBase):
    __tablename__ = 'module_group_member'
    description = mapped_column(String)
    is_active = mapped_column(Boolean)
    module_pk = mapped_column(ForeignKey('module.pk'), primary_key=True,)
    module_group_pk = mapped_column(ForeignKey('module_group.pk'), primary_key=True)

执行查询整个Module对象的语句时，可正常获取关联的groups：

query = select(Module).where(Module.name == 'Module3')  # type: ignore
results = session.execute(query).unique().all()
print('results', results)

但执行仅查询Module.name和Module.groups的语句时，出现笛卡尔积警告：

query = select(Module.name, Module.groups).where(Module.name == 'Module3') # type: ignore

SAWarning: SELECT statement has a cartesian product between FROM element(s) "module_group", "module_group_member_1" and FROM element "module". Apply join condition(s) between each element to resolve.

需求是实现获取所有模块名称及其关联组的查询：

query = select(Module.name, Module.groups)

原因分析 #

当直接选择实体的单个属性（如Module.name）和关联集合（如Module.groups）时，SQLAlchemy不会自动利用relationship配置的关联规则构建正确的JOIN语句，而是直接将三张表做无关联连接，从而产生笛卡尔积警告。而查询整个Module实体时，SQLAlchemy会根据relationship的secondary参数自动处理表关联，避免笛卡尔积。

解决方案 #

方案1：显式构建JOIN语句 #

通过join方法手动关联中间表和ModuleGroup，确保关联条件正确：

from sqlalchemy import select

# 查询单个模块的名称和关联组
query = (
    select(Module.name, ModuleGroup)
    .join(ModuleGroupMember, Module.pk == ModuleGroupMember.module_pk)
    .join(ModuleGroup, ModuleGroup.pk == ModuleGroupMember.module_group_pk)
    .where(Module.name == 'Module3')
)
results = session.execute(query).all()

# 处理结果，聚合同一模块的组
module_groups = {}
for name, group in results:
    if name not in module_groups:
        module_groups[name] = []
    module_groups[name].append(group)

方案2：使用聚合函数分组查询 #

利用func.array_agg（适用于PostgreSQL）或类似聚合函数，将每个模块的关联组聚合为列表，避免笛卡尔积：

from sqlalchemy import select, func

# 查询所有模块名称及其关联组（PostgreSQL示例）
query = (
    select(
        Module.name,
        func.array_agg(ModuleGroup.name).label('group_names')
    )
    .outerjoin(ModuleGroupMember, Module.pk == ModuleGroupMember.module_pk)
    .outerjoin(ModuleGroup, ModuleGroup.pk == ModuleGroupMember.module_group_pk)
    .group_by(Module.name)
)
results = session.execute(query).all()

# 结果格式：[(模块名称, [组名1, 组名2, ...]), ...]

方案3：查询完整实体后提取字段 #

如果数据量不大，可继续查询完整Module实体，在Python层面提取所需字段：

query = select(Module)
results = session.execute(query).scalars().all()

module_info = [(module.name, module.groups) for module in results]

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