SQLAlchemy ORM中CTEs列的访问与aliased该如何应用

SQLAlchemy ORM中CTEs的基础定义

CTEs即公用表表达式，是SQL中用于简化复杂查询的功能，在SQLAlchemy ORM中可以通过cte()方法将查询定义为CTE。首先我们需要构建基础的模型，这里以用户和订单两个常见实体为例，定义对应的ORM模型。

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

# 创建基类
Base = declarative_base()

# 定义用户模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    # 关联订单
    orders = relationship("Order", back_populates="user")

# 定义订单模型
class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    order_no = Column(String(50))
    user_id = Column(Integer, ForeignKey('users.id'))
    # 关联用户
    user = relationship("User", back_populates="orders")

# 初始化数据库连接和会话
engine = create_engine('sqlite:///test.db')
Session = sessionmaker(bind=engine)
session = Session()

CTEs列的访问方式

当我们通过cte()方法将查询转换为CTE后，访问其中的列不能直接使用原模型的属性，需要通过CTE对象的c属性来获取。下面定义一个查询所有用户订单数量的CTE，演示列的访问操作。

from sqlalchemy import func

# 定义查询用户订单数量的子查询
user_order_count_query = session.query(
    User.id.label('user_id'),
    User.name.label('user_name'),
    func.count(Order.id).label('order_count')
).join(Order, User.id == Order.user_id).group_by(User.id, User.name)

# 将子查询转换为CTE
user_order_cte = user_order_count_query.cte(name='user_order_cte')

# 访问CTE的列，通过c属性获取
print(user_order_cte.c.user_id)
print(user_order_cte.c.user_name)
print(user_order_cte.c.order_count)

# 查询CTE中的数据
results = session.query(user_order_cte).all()
for row in results:
    print(f"用户ID: {row.user_id}, 用户名: {row.user_name}, 订单数量: {row.order_count}")

aliased在CTEs中的应用场景

场景一：解决CTE多次引用的别名冲突

当同一个CTE需要在查询中多次引用时，直接重复使用该CTE对象会导致SQL语句中的表名冲突，此时可以使用aliased为CTE创建不同的别名，避免冲突。比如我们需要查询订单数量大于平均订单数量的用户，就需要两次引用用户订单CTE。

from sqlalchemy.orm import aliased

# 为user_order_cte创建两个别名
cte_alias1 = aliased(user_order_cte, name='cte_alias1')
cte_alias2 = aliased(user_order_cte, name='cte_alias2')

# 查询订单数量大于平均订单数量的用户
avg_order_count = session.query(func.avg(cte_alias1.c.order_count)).scalar()
target_users = session.query(cte_alias2).filter(cte_alias2.c.order_count > avg_order_count).all()

print(f"平均订单数量: {avg_order_count}")
for user in target_users:
    print(f"用户ID: {user.user_id}, 用户名: {user.user_name}, 订单数量: {user.order_count}")

场景二：CTE与其他表关联时的别名处理

当CTE需要和用户表或者其他模型表进行关联查询时，使用aliased可以更清晰地管理关联逻辑，同时也能避免列名冲突的问题。下面的示例演示如何将用户订单CTE和用户表进行关联，获取用户的详细信息。

from sqlalchemy.orm import aliased

# 为用户表创建别名，为CTE创建别名
user_alias = aliased(User, name='user_alias')
cte_alias = aliased(user_order_cte, name='cte_alias')

# 关联查询用户表和CTE
join_results = session.query(
    user_alias.id,
    user_alias.name,
    cte_alias.c.order_count
).join(cte_alias, user_alias.id == cte_alias.c.user_id).all()

for row in join_results:
    print(f"用户ID: {row.id}, 用户名: {row.name}, 订单数量: {row.order_count}")

注意事项

• 访问CTE的列时，必须使用CTE对象的c属性，不能直接使用原模型的属性，否则会抛出属性错误。
• 使用aliased为CTE创建别名时，建议指定有意义的name参数，这样生成的SQL语句可读性更强，也方便后续调试。
• CTE的定义依赖底层的查询语句，如果底层查询的列没有通过label指定别名，访问列时需要使用默认的列名，容易出现混淆，建议对CTE中的列统一指定清晰的别名。
• 不同的数据库对CTEs的支持程度略有差异，在使用时需要注意当前数据库的SQL语法规范，避免出现兼容性问题。

SQLAlchemyORMCTEsaliased修改时间：2026-07-08 21:42:12