如何在SQLAlchemy的ORM层面实现自关联完整链条的预加载

针对PostgreSQL环境下的自关联Node模型，要实现一次查询递归加载完整的prev关联链条（包括Node作为其他模型关联对象的场景），可以通过以下几种方式实现：

方法一：自定义查询函数，一次性加载全链条 #

通过构造递归CTE获取所有关联节点的ID，批量加载这些节点到Session中，SQLAlchemy会自动关联prev属性：

from sqlalchemy import select
from sqlalchemy.orm import Session

def get_node_with_full_prev_chain(session: Session, node_id: int) -> Node:
    # 构建递归CTE，获取当前节点及所有上游prev节点的ID
    cte = select(Node.id).where(Node.id == node_id).cte(recursive=True)
    cte = cte.union_all(
        select(Node.id).join(cte, Node.id == cte.c.prev_id)
    )
    # 批量加载所有关联节点到Session
    all_nodes = session.query(Node).filter(Node.id.in_(select(cte.c.id))).all()
    # 返回目标节点
    return next(node for node in all_nodes if node.id == node_id)

使用示例：

target_node = get_node_with_full_prev_chain(session, 123)
# 直接访问完整链条，无额外查询
current = target_node
while current:
    print(current.id)
    current = current.prev

方法二：自定义懒加载器，访问时自动触发递归加载 #

修改prev关系的懒加载逻辑，第一次访问prev时自动加载全链条：

from sqlalchemy import select
from sqlalchemy.orm import object_session, LoaderCallableStatus

def recursive_prev_loader(_, instance, status: LoaderCallableStatus):
    # 已加载或无上级节点时直接返回
    if status.loaded or instance.prev_id is None:
        return
    session = object_session(instance)
    # 递归CTE获取全链条节点ID
    cte = select(Node.id).where(Node.id == instance.id).cte(recursive=True)
    cte = cte.union_all(
        select(Node.id).join(cte, Node.id == cte.c.prev_id)
    )
    # 批量加载所有节点到Session
    session.query(Node).filter(Node.id.in_(select(cte.c.id))).all()
    # 标记为已加载，避免重复查询
    status.loaded = True

# 更新Node模型的prev关系定义
class Node(Base):
    __tablename__ = 'node'
    id = Column(Integer(), primary_key=True)
    prev_id = Column(Integer(), ForeignKey(id))
    prev = relationship('Node', uselist=False, lazy=recursive_prev_loader)

效果：第一次访问任意Node的prev属性时，会自动加载整个上游链条，后续访问无需额外查询。

方法三：关联其他模型时自动加载全链条 #

如果Node是其他模型的关联对象（如OtherModel），可以在查询关联模型时，附加递归加载逻辑：

from sqlalchemy.orm import selectinload

# 假设关联模型定义
class OtherModel(Base):
    __tablename__ = 'other'
    id = Column(Integer(), primary_key=True)
    node_id = Column(Integer(), ForeignKey('node.id'))
    node = relationship('Node')

def get_other_with_full_node_chain(session: Session, other_id: int) -> OtherModel:
    target_other = session.query(OtherModel).filter(OtherModel.id == other_id).first()
    if target_other.node:
        # 加载关联Node的全链条
        cte = select(Node.id).where(Node.id == target_other.node.id).cte(recursive=True)
        cte = cte.union_all(
            select(Node.id).join(cte, Node.id == cte.c.prev_id)
        )
        session.query(Node).filter(Node.id.in_(select(cte.c.id))).all()
    return target_other

原理说明：
PostgreSQL递归CTE一次性获取所有关联节点ID后，SQLAlchemy Session会自动管理对象引用——只要prev_id对应的Node已在Session中，prev属性就会自动指向该对象，无需额外JOIN操作，完美实现一次查询加载全链条的需求。

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