SQLAlchemy AsyncSession查询报错：需调用Result.unique()方法求因

问题背景 #

模型类定义 #

import enum
import uuid
from sqlalchemy import Column, UUID, String, Boolean, DateTime, ForeignKey, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from typing import List, Any

Base = declarative_base()

class PricingFrequency(enum.Enum):
    month = 'month'
    year = 'year'


class PlanPricing(Base):
    __tablename__ = "PlansPricing"
    pricing_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    # 省略其他字段
    subscription_plan = relationship("SubscriptionPlan", back_populates="plans_pricing")
    plan_id = Column(UUID(as_uuid=True), ForeignKey("SubscriptionPlans.plan_id"))

    created_on = Column(DateTime, server_default=func.now())
    updated_on = Column(DateTime, server_default=func.now(), server_onupdate=func.now())


class SubscriptionPlanOption(Base):
    __tablename__ = "SubscriptionPlanOptions"
    option_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    # 省略其他字段
    subscription_plan = relationship("SubscriptionPlan", back_populates="options_plan")
    plan_id = Column(UUID(as_uuid=True), ForeignKey("SubscriptionPlans.plan_id"))

    created_on = Column(DateTime, server_default=func.now())
    updated_on = Column(DateTime, server_default=func.now(), server_onupdate=func.now())


class SubscriptionPlan(Base):
    __tablename__ = "SubscriptionPlans"
    plan_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    plan_name = Column(String)
    plan_description = Column(String)
    is_popular = Column(Boolean, default=False)

    plans_pricing: List[Any] = relationship("PlanPricing", back_populates="subscription_plan")
    options_plan: List[Any] = relationship("SubscriptionPlanOption", back_populates="subscription_plan")

    created_on = Column(DateTime, server_default=func.now())
    updated_on = Column(DateTime, server_default=func.now(), server_onupdate=func.now())

查询代码 #

# 假设pricing_id是已定义的变量
query = (
    select(SubscriptionPlan)
    .options(
        joinedload(SubscriptionPlan.options_plan, innerjoin=True),
        joinedload(SubscriptionPlan.plans_pricing.and_(PlanPricing.pricing_id == pricing_id), innerjoin=True)
    )
)
items = await session.execute(query)
items = items.scalars().all()

错误信息 #

The unique() method must be invoked on this Result, as it contains results that include joined eager loads against collections

问题根源 #

当使用joinedload同时加载两个或多个集合类型的关联关系（这里options_plan和plans_pricing都是一对多关联）时，SQL查询会生成笛卡尔积结果：同一个SubscriptionPlan实例会对应多行数据库记录（行数等于该Plan关联的Pricing数量 × Option数量）。

SQLAlchemy的scalars()方法默认不会自动处理重复的实体实例，因为它无法判断是否需要保留重复结果。为了确保返回的每个SubscriptionPlan实例唯一，必须显式调用unique()方法，让SQLAlchemy根据实体主键识别并合并重复的实体对象。

解决方法 #

在调用scalars()之前添加unique()方法：

items = await session.execute(query)
items = items.unique().scalars().all()

unique()会基于实体主键对结果去重，确保每个SubscriptionPlan只返回一次，同时正确加载对应的关联集合数据。

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