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

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, select
from sqlalchemy.orm import sessionmaker, declarative_base, aliased

# 基础声明
Base = declarative_base()

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

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

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

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

# 数据库设置 (仅为示例，使用内存SQLite)
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 填充数据
session.add_all([
    User(id=1, name='Alice', email_address='alice@example.com'),
    User(id=2, name='Bob', email_address='bob@example.com'),
    Transaction(txn_id=101, user_id=1, product_id=10),
    Transaction(txn_id=102, user_id=1, product_id=20),
    Transaction(txn_id=103, user_id=2, product_id=30),
])
session.commit()

# 创建一个简单的 CTE
user_cte = (
    select(User)
    .where(User.email_address == 'alice@example.com')
    .cte()
)

# 错误示例：直接访问 CTE 对象属性
# select(user_cte.id) # 这会抛出 AttributeError: CTE object has no attribute 'id'

# 正确示例：通过 .c 属性访问 CTE 列
# 访问用户ID
result = session.execute(select(user_cte.c.id)).scalar_one_or_none()
print(f"通过 CTE 访问到的用户ID: {result}") # 输出: 通过 CTE 访问到的用户ID: 1

# 访问用户名称
result_name = session.execute(select(user_cte.c.name)).scalar_one_or_none()
print(f"通过 CTE 访问到的用户名称: {result_name}") # 输出: 通过 CTE 访问到的用户名称: Alice

# 创建一个包含 User 和 Transaction 连接的 CTE
user_transactions_cte = (
    select(User.id, User.name, Transaction.txn_id, Transaction.product_id)
    .join(Transaction, User.id == Transaction.user_id)
    .where(User.email_address == 'alice@example.com')
    .cte()
)

# 现在，我们可以通过 .c 属性访问 CTE 中的所有列
print("\n--- 访问多表连接 CTE 的列 ---")
# 访问用户ID和交易ID
results = session.execute(
    select(user_transactions_cte.c.id, user_transactions_cte.c.name, user_transactions_cte.c.txn_id, user_transactions_cte.c.product_id)
).all()

for row in results:
    print(f"用户ID: {row.id}, 姓名: {row.name}, 交易ID: {row.txn_id}, 产品ID: {row.product_id}")
# 输出：
# 用户ID: 1, 姓名: Alice, 交易ID: 101, 产品ID: 10
# 用户ID: 1, 姓名: Alice, 交易ID: 102, 产品ID: 20

# CTE 只选择 User 的所有列
user_only_cte = (
    select(User)
    .where(User.name == 'Bob')
    .cte()
)

# 使用 aliased 将 CTE 映射到 User 类
AliasedUser = aliased(User, user_only_cte)

# 现在可以像操作 User 对象一样操作 AliasedUser
print("\n--- 使用 aliased 映射 CTE ---")
bob_user = session.execute(select(AliasedUser)).scalar_one_or_none()
if bob_user:
    print(f"通过 aliased 映射的 User 对象: {bob_user.name}, {bob_user.email_address}")
# 输出：通过 aliased 映射的 User 对象: Bob, bob@example.com