SQLAlchemy中使用column_property实现递归属性时出现列数不匹配CompileError的解决咨询

我看到你在尝试用SQLAlchemy的column_property结合递归CTE来实现获取分类的所有深层子节点，但遇到了列数不匹配的编译错误，这个问题其实是递归CTE的初始锚点查询和后续递归查询的列数不一致导致的，同时还有个小问题是你的CTE没有关联到当前分类实例，我来帮你一步步解决这个问题。

一、错误原因分析 #

你当前的递归CTE存在两个关键问题：

1. 列数不匹配：
   • 初始锚点查询是select(Category.id)，只返回1列（id）
   • 而递归部分的查询是select(Category)，这会返回Category表的所有列（id、parent_id），两者列数不同，直接导致SQLAlchemy抛出CompileError。
2. 未关联当前实例：
   • 你的CTE没有和当前的Category实例绑定，最终生成的deep_children会是全局的所有分类，而不是当前分类的专属后代节点。

二、修正方案 #

我们需要调整递归CTE的逻辑，保证锚点查询和递归部分的列数、类型完全一致，同时让CTE关联到当前分类实例的id，这样每个分类实例的deep_children就能返回自己的深层子节点。

1. 确定deep_children的预期返回值 #

首先明确你需要deep_children返回什么：

• 如果是所有深层子节点的ID列表（PostgreSQL支持数组聚合）
• 如果是深层子节点的数量

下面分别给出两种场景的修正代码：

2. 修正后的代码示例 #

场景1：返回所有深层子节点的ID列表

from sqlalchemy import (
    Column,
    ForeignKey,
    Integer,
    create_engine,
    select,
    func
)
from sqlalchemy.orm import (
    backref,
    column_property,
    declarative_base,
    relationship,
    remote,
    Session,
)
from testcontainers.postgres import PostgresContainer


Base = declarative_base()


class Category(Base):
    __tablename__ = "categories"

    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey("categories.id"), index=True, nullable=True)

    children = relationship(
        "Category",
        primaryjoin=(id == remote(parent_id)),
        lazy="select",
        backref=backref(
            "parent",
            primaryjoin=(remote(id) == parent_id),
            lazy="select",
        ),
    )


def build_deep_children_expr():
    # 锚点CTE：以当前分类的ID作为起始（关联当前实例的id）
    cte = select(
        Category.id.label("descendant_id")
    ).cte(recursive=True, name="descendants_cte")
    
    # 递归部分：查询父ID等于CTE中descendant_id的分类ID，保证列数和锚点一致
    recursive_part = select(
        cat.id.label("descendant_id")
    ).select_from(Category.__table__.alias("cat")).where(
        cat.parent_id == cte.c.descendant_id
    )
    
    # 合并锚点和递归查询
    cte = cte.union_all(recursive_part)
    
    # 聚合所有后代ID，排除当前分类自身（锚点包含了自己）
    return select(func.array_agg(cte.c.descendant_id)).where(
        cte.c.descendant_id != Category.id
    ).scalar_subquery()


# 为Category类绑定deep_children属性
Category.deep_children = column_property(build_deep_children_expr())

if __name__ == "__main__":
    with PostgresContainer("postgres:16.1") as postgres:
        database_url = postgres.get_connection_url(driver="psycopg")
        engine = create_engine(database_url, echo=True)

        Base.metadata.create_all(engine)

        session = Session(engine)

        # 构建分类层级
        root = Category(
            id=0,
            children=[
                Category(id=1),
                Category(
                    id=2,
                    children=[
                        Category(id=3),
                        Category(id=4),
                        Category(id=5, children=[Category(id=6), Category(id=7)]),
                    ],
                ),
            ],
        )

        session.add(root)
        session.flush()
        session.expire_all()

        cat = session.query(Category).filter(Category.id == 0).one()

        print(f"Category: {cat!r}")
        print(f"  deep_children (ID列表): {cat.deep_children}")
        print(f"  直接子节点: {[c.id for c in cat.children]}")

        # 检查每个子节点的深层子节点
        for c in cat.children:
            print(f"    子节点 {c.id} 的深层子节点: {c.deep_children}")

        session.close()

场景2：返回深层子节点的数量

如果只需要子节点数量，只需修改build_deep_children_expr的最后部分：

def build_deep_children_expr():
    # 锚点CTE：以当前分类的ID作为起始（关联当前实例的id）
    cte = select(
        Category.id.label("descendant_id")
    ).cte(recursive=True, name="descendants_cte")
    
    # 递归部分：查询父ID等于CTE中descendant_id的分类ID，保证列数和锚点一致
    recursive_part = select(
        cat.id.label("descendant_id")
    ).select_from(Category.__table__.alias("cat")).where(
        cat.parent_id == cte.c.descendant_id
    )
    
    # 合并锚点和递归查询
    cte = cte.union_all(recursive_part)
    
    # 统计后代数量，排除当前分类自身
    return select(func.count(cte.c.descendant_id)).where(
        cte.c.descendant_id != Category.id
    ).scalar_subquery()

三、关键注意事项 #

• 递归CTE的列一致性：锚点查询和递归部分的查询必须返回相同数量、相同类型的列，这是SQL递归CTE的硬性要求。
• 关联当前实例：通过在锚点查询中使用Category.id，我们让CTE和当前的分类实例绑定，确保每个实例的deep_children只返回自己的后代。
• lazy加载特性：因为column_property是在主查询时直接计算的，所以它会随主查询一起加载，不会像relationship那样有延迟加载的行为。

备注：内容来源于stack exchange，提问作者guhur