SQLAlchemy自引用模型混合属性表达式实现求助

我有一个自引用的Node模型，其中ref_id是指向自身的外键，可通过associated_refs属性访问父节点的子节点集合：

import enum
from sqlalchemy import Column, Integer, Enum, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref

Base = declarative_base()

class Node(Base):

    class NodeStatus(enum.IntEnum):
        STATUS_1 = 0
        STATUS_2 = 1

    id = Column(Integer, primary_key=True)
    status = Column(Enum(NodeStatus))
    ref_id = Column(Integer, ForeignKey("Node.id", ondelete="SET NULL"))

    ref = relationship(
        "Node",
        foreign_keys=[ref_id],
        uselist=False,
        remote_side="Node.id",
        backref=backref("associated_refs")
    )

我想要定义一个混合属性，用于判断该实例的父节点状态是否为STATUS_1，已经实现了实例方法，但不知道如何编写对应的expression方法：

from sqlalchemy.ext.hybrid import hybrid_property

class Node(Base):
    # 上述模型代码...
    @hybrid_property
    def ref_is_status_1(self) -> bool:
        if self.associated_refs:
            return self.associated_refs[0].status == self.NodeStatus.STATUS_1
        else:
            return False

    @ref_is_status_1.expression
    def ref_is_status_1(cls):
        # 此处需要实现
        pass

要求无需通过SQLAlchemy的session.execute进行查询，保证可测试性和性能（数据库中有数十万节点）。我尝试过两种写法均失败：

• 使用count的表达式，始终返回空结果：

from sqlalchemy import func, and_

@ref_is_status_1.expression
def ref_is_status_1(cls):
    return select(func.count(Node.id)).where(
                 and_(
                     Node.ref_id == cls.id,
                     Node.status == Node.NodeStatus.STATUS_1
                 )).label("ref_is_status_1")

• 使用exists和case的写法，抛出ProgrammingError错误（提示未指定表的SELECT *语句）：

from sqlalchemy import exists, select, case, and_

@ref_is_status_1.expression
def ref_is_status_1(cls):
    return (
        select([
            case([(exists().where(
                and_(
                    Node.ref_id == cls.id,
                    Node.status == Node.NodeStatus.STATUS_1
                )).correlate(cls), True)],
                 else_=False
                 ).label("has_ref_is_status_1")
        ]).label("ref_is_status_1")
    )

1. 修正混合属性的实例方法 #

首先要纠正关联关系的逻辑错误：associated_refs是父节点的子节点集合，当前实例的父节点应该通过self.ref访问，而不是取associated_refs[0]。正确的实例方法如下：

@hybrid_property
def ref_is_status_1(self) -> bool:
    return self.ref is not None and self.ref.status == self.NodeStatus.STATUS_1

2. 实现对应的expression方法 #

使用exists子查询来判断是否存在符合条件的父节点，关联条件要匹配当前节点的ref_id等于父节点的id，同时父节点状态为STATUS_1：

from sqlalchemy import exists, select, and_

@ref_is_status_1.expression
def ref_is_status_1(cls):
    return exists(
        select(1).where(
            and_(
                Node.id == cls.ref_id,
                Node.status == cls.NodeStatus.STATUS_1
            )
        )
    ).label("ref_is_status_1")

错误原因分析 #

• count写法错误：关联条件搞反了，Node.ref_id == cls.id是查询当前节点的子节点，而非父节点，逻辑完全不符合需求，导致返回空结果。
• exists+case写法错误：同样关联条件逻辑颠倒，且无需嵌套select和case，SQLAlchemy会自动将exists表达式转换为布尔值，多余的嵌套导致语法错误。

性能验证 #

该写法生成的EXISTS子查询可以利用id和ref_id上的索引（建议为这两个字段添加索引），数十万数据量下性能表现良好，可直接在查询中使用该混合属性，例如：

# 查询所有父节点状态为STATUS_1的节点
nodes = session.query(Node).filter(Node.ref_is_status_1).all()

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