SQLAlchemy ORM中CTE与别名的高效使用及列访问指南

import sqlalchemy as sa
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base, aliased, relationship

# 声明基类
Base = declarative_base()

# 定义User模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    email_address = Column(String, unique=True, nullable=False)
    name = Column(String)

    transactions = relationship("Transaction", back_populates="user")

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email_address}')>"

# 定义Transaction模型
class Transaction(Base):
    __tablename__ = 'transactions'
    txn_id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    product_id = Column(Integer)

    user = relationship("User", back_populates="transactions")

    def __repr__(self):
        return f"<Transaction(txn_id={self.txn_id}, user_id={self.user_id}, product_id={self.product_id})>"

# 数据库连接与会话（此处仅为示例，实际使用时需配置）
# engine = create_engine('sqlite:///:memory:')
# Base.metadata.create_all(engine)
# Session = sessionmaker(bind=engine)
# session = Session()

# 示例：创建一个筛选特定email用户的CTE
user_email = 'test@example.com'
user_cte_query = (
    sa.select(User)
    .where(User.email_address == user_email)
    .cte("filtered_users_cte") # 为CTE命名
)

# 使用aliased将CTE映射回User ORM类
AliasedUserFromCTE = aliased(User, user_cte_query)

# 现在可以像访问User对象一样访问AliasedUserFromCTE的属性
# 例如，查询CTE中用户的ID和名称
query_user_from_cte = sa.select(AliasedUserFromCTE.id, AliasedUserFromCTE.name)
print("查询AliasedUserFromCTE的ID和名称的SQL：")
print(query_user_from_cte.compile(compile_kwargs={"literal_binds": True}))
# 预期输出类似：SELECT filtered_users_cte.id, filtered_users_cte.name FROM users AS filtered_users_cte WHERE filtered_users_cte.email_address = 'test@example.com'

# 假设我们有一个CTE，直接尝试访问其列会报错
# select(user_cte_query.id) # 这会抛出AttributeError: 'CTE' object has no attribute 'id'

# 正确的方法是使用.c属性
query_user_id_direct = sa.select(user_cte_query.c.id)
print("\n直接通过.c访问CTE的ID的SQL：")
print(query_user_id_direct.compile(compile_kwargs={"literal_binds": True}))
# 预期输出类似：SELECT filtered_users_cte.id FROM users AS filtered_users_cte WHERE filtered_users_cte.email_address = 'test@example.com'

# 创建一个包含User和Transaction联接的CTE
multi_table_join_cte = (
    sa.select(User.id, User.name, Transaction.txn_id)
    .join(Transaction, User.id == Transaction.user_id)
    .where(User.email_address == user_email)
    .cte("user_transactions_cte")
)

# 访问多表JOIN CTE中的列
# 正确方法：使用.c属性来访问所有投影的列
query_multi_table_cte = sa.select(
    multi_table_join_cte.c.id,
    multi_table_join_cte.c.name,
    multi_table_join_cte.c.txn_id
)
print("\n查询多表JOIN CTE的ID、名称和交易ID的SQL：")
print(query_multi_table_cte.compile(compile_kwargs={"literal_binds": True}))
# 预期输出类似：SELECT user_transactions_cte.id, user_transactions_cte.name, user_transactions_cte.txn_id FROM users JOIN transactions ON users.id = transactions.user_id WHERE users.email_address = 'test@example.com'