如何用with_expression()从关联表填充多对多关系中User实例的position？

问题说明 #

现有User和Group两个实体，通过带position字段的UserGroup关联表实现多对多关系。需求是查询Group时，User实体能直接获取到其在该组中的position值（即通过group.users[0].position访问），但使用query_expression()配合with_expression()时，User.position始终为None。

原因分析 #

原查询中使用joinedload(UserGroup.user)属于独立加载逻辑，SQLAlchemy无法将UserGroup.position与User的query_expression字段建立关联——因为joinedload会生成独立子查询，导致UserGroup和User的关联上下文丢失。而literal()能生效是因为它是独立常量，不需要依赖关联上下文。

解决方案 #

使用contains_eager替代joinedload，明确将User与当前查询中的UserGroup关联起来，让SQLAlchemy能正确映射UserGroup.position到User.position。

修改后的查询代码 #

group = session.execute(
    select(Group).options(
        selectinload(Group.group_users)
        .contains_eager(UserGroup.user)
        .with_expression(User.position, UserGroup.position)
    ).where(Group.id == 1)
).scalar_one()

完整可运行代码 #

from sqlalchemy import create_engine, literal, select, ForeignKey
from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy
from sqlalchemy.orm import (
    DeclarativeBase, declared_attr, Mapped, mapped_column, query_expression,
    relationship, selectinload, contains_eager, Session
)


class Base(DeclarativeBase):
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower()


class User(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    user_groups: Mapped[list['UserGroup']] = relationship(
        'UserGroup', back_populates='user'
    )
    position: Mapped[str | None] = query_expression()

    def __repr__(self) -> str:
        return f'User(id={self.id}, name={self.name})'


class Group(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column()
    group_users: Mapped[list['UserGroup']] = relationship(
        'UserGroup', back_populates='group'
    )
    users: AssociationProxy[list['User']] = association_proxy(
        target_collection='group_users', attr='user'
    )

    def __repr__(self) -> str:
        return f'Group(id={self.id}, title={self.title})'


class UserGroup(Base):
    user_id: Mapped[int] = mapped_column(
        ForeignKey("user.id"), primary_key=True
    )
    group_id: Mapped[int] = mapped_column(
        ForeignKey("group.id"), primary_key=True
    )
    position: Mapped[str] = mapped_column()

    group = relationship('Group', back_populates='group_users')
    user = relationship('User', back_populates='user_groups')


# 初始化数据库和测试数据
engine = create_engine("sqlite+pysqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)

user_1 = User(name="Bob")
user_2 = User(name="Ben")
group_1 = Group(title="Editors")
position_1 = UserGroup(user=user_1, group=group_1, position="Senior")
position_2 = UserGroup(user=user_2, group=group_1, position="Junior")

session = Session(bind=engine)
session.add_all([user_1, user_2, group_1, position_1, position_2])
session.commit()

# 修改后的查询
group = session.execute(
    select(Group).options(
        selectinload(Group.group_users)
        .contains_eager(UserGroup.user)
        .with_expression(User.position, UserGroup.position)
    ).where(Group.id == 1)
).scalar_one()

print(f'Members of group {group.title}:')
for user in group.users:
    print(f'User.name={user.name}, User.position={user.position}')

运行结果 #

Members of group Editors:
User.name=Bob, User.position=Senior
User.name=Ben, User.position=Junior

补充说明 #

• contains_eager用于显式关联已在主查询中连接的实体，它不像joinedload那样生成独立查询，而是复用当前查询的关联上下文，因此能正确传递UserGroup.position的值到User的query_expression字段。
• 如果同一个用户属于多个组，这种方式在查询单个组时只会填充该组对应的position，符合每个User-Group对关联记录唯一的需求。

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