如何编写依赖父查询表的SQLAlchemy子查询？

现有SQLAlchemy模型 #

class BaseModel( DeclarativeBase ):
    pass

class ABC( BaseModel ):

    __tablename__   = "ABC"
    __table_args__  = (
        Index( "index1", "account_id" ),
        ForeignKeyConstraint( [ "account_id" ], [ "A.id" ], onupdate = "CASCADE", ondelete = "CASCADE" ),
    )

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

class A( BaseModel ):

    __tablename__   = "A"
    __table_args__  = (
        Index( "index1", "downloaded", "idx" ),
    )

    id:             Mapped[ int ]   = mapped_column( primary_key = True, autoincrement = True )
    downloaded:     Mapped[ date ]
    account_id:     Mapped[ str ]   = mapped_column( String( 255 ) )
    display_name:   Mapped[ str ]   = mapped_column( String( 255 ) )
    idx:            Mapped[ int ]
    type:           Mapped[ str ]   = mapped_column( String( 255 ) )

目标原生SQL #

select
    a.account_id,
    (
        select
            group_concat( a2.account_id )
        from
            ABC abc
            left join A a2 on a2.downloaded = a.downloaded and a2.idx = abc.account_idx
        where
            abc.account_id = a.id
    ) as 'brokerage_client_accounts'
from
    A a
where
    a.downloaded = "2024-11-12" and
    a.type != 'SYSTEM'
order by
    a.account_id
;

错误代码及报错信息 #

尝试的代码 #

A2 = aliased( A )

brokerage_client_accounts_subq = select(
    func.aggregate_strings( A2.account_id, "," ).label( "accounts" ),
).select_from(
    A
).outerjoin(
    A2,
    and_( A2.downloaded == A.downloaded, A2.idx == ABC.account_idx )
).where(
    ABC.account_id == A.id
)

stmt = select(
    Account.account_id,
    brokerage_client_accounts_subq.c.accounts,
).where(
    and_(
        A.downloaded == date( 2024, 11, 12 ),
        A.type != "SYSTEM"
    )
).order_by(
    Account.account_id
)

报错信息 #

SAWarning: SELECT statement has a cartesian product between FROM element(s) "anon_1" and FROM element "A".  Apply join condition(s) between each element to resolve.

mysql.connector.errors.ProgrammingError: 1054 (42S22): Unknown column 'ABC.account_idx' in 'on clause'

修正后的代码及解释 #

核心错误原因 #

1. 子查询FROM子句错误：未引入ABC表，导致字段无法识别
2. 误用聚合函数：MySQL字符串聚合用group_concat而非aggregate_strings
3. 关联逻辑错误：子查询应作为关联子查询直接引用父表字段，而非在子查询FROM中重复引入父表
4. 模型引用错误：主查询中误用Account（实际应为A）

修正代码 #

from sqlalchemy import select, func, aliased, and_
from datetime import date

A2 = aliased(A)

# 构建关联子查询，直接引用父查询的A表字段
brokerage_client_accounts_subq = select(
    func.group_concat(A2.account_id).label("brokerage_client_accounts")
).select_from(ABC).outerjoin(
    A2,
    and_(
        A2.downloaded == A.downloaded,
        A2.idx == ABC.account_idx
    )
).where(ABC.account_id == A.id)

# 主查询
stmt = select(
    A.account_id,
    brokerage_client_accounts_subq.label("brokerage_client_accounts")
).where(
    and_(
        A.downloaded == date(2024, 11, 12),
        A.type != "SYSTEM"
    )
).order_by(A.account_id)

关键调整说明 #

• 子查询select_from改为ABC，匹配原生SQL的表关联顺序，确保ABC字段可用
• 使用func.group_concat适配MySQL的聚合语法
• 子查询直接引用父查询的A.downloaded和A.id，形成关联子查询，避免笛卡尔积警告
• 主查询统一使用A模型，修正引用错误
• 将子查询通过label嵌入主查询SELECT列表，完全对齐目标原生SQL结构

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