如何用SQLAlchemy实现带WITH子句的指定SQL查询？

你的问题核心是错误地将原表与CTE同时传入select_from方法，导致生成冗余的FROM子句。正确的实现逻辑是先定义CTE对象，再基于该CTE构建后续的关联查询，具体实现如下：

基于SQLAlchemy Core的实现 #

from sqlalchemy import create_engine, select, MetaData, Table, Column, Integer, String, DateTime

# 模拟表结构（请根据实际项目调整）
metadata = MetaData()
user_table = Table(
    "user", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("status", String),
    Column("created_at", DateTime),
    Column("account_id", Integer)
)

account_table = Table(
    "account", metadata,
    Column("id", Integer, primary_key=True)
)

# 1. 定义CTE
filtered_users_cte = (
    select(
        user_table.c.id,
        user_table.c.name,
        user_table.c.status,
        user_table.c.account_id  # 关联account表需要该字段，必须包含
    )
    .where(user_table.c.status == "ACTIVE")
    .order_by(user_table.c.created_at.desc())
    .limit(100)
    .offset(0)
).cte("filtered_users")

# 2. 构建主查询
query = (
    select(
        filtered_users_cte.c.id,
        filtered_users_cte.c.name,
        filtered_users_cte.c.status,
        account_table.c.id.label("id_2")
    )
    .select_from(filtered_users_cte)  # 仅传入CTE，不传入原user表
    .outerjoin(
        account_table,
        filtered_users_cte.c.account_id == account_table.c.id  # 用CTE的字段做关联条件
    )
)

# 执行示例
engine = create_engine("your_database_connection_url")
with engine.connect() as conn:
    result = conn.execute(query)
    for row in result:
        print(row)

基于SQLAlchemy ORM的实现（使用模型类） #

假设你已定义ORM模型User和Account：

from sqlalchemy.orm import sessionmaker
from your_module import User, Account  # 导入你的模型类

# 1. 定义CTE
filtered_users_cte = (
    select(
        User.id,
        User.name,
        User.status,
        User.account_id
    )
    .where(User.status == "ACTIVE")
    .order_by(User.created_at.desc())
    .limit(100)
    .offset(0)
).cte("filtered_users")

# 2. 构建主查询
query = (
    select(
        filtered_users_cte.c.id,
        filtered_users_cte.c.name,
        filtered_users_cte.c.status,
        Account.id.label("id_2")
    )
    .select_from(filtered_users_cte)
    .outerjoin(Account, filtered_users_cte.c.account_id == Account.id)
)

# 执行示例
Session = sessionmaker(bind=engine)
with Session() as session:
    result = session.execute(query).all()

关键注意事项 #

• 定义CTE时必须包含后续关联所需的字段（比如account_id），否则无法与account表建立关联。
• 主查询的select_from仅传入CTE对象，不要同时传入原user表，这是避免冗余FROM子句的核心。
• 关联条件必须使用CTE的字段，而非原user表的字段，否则会导致额外的表引用或查询错误。

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