SQLAlchemy递归表查询优化求助：Artifact表CRUD性能提升

刚接触SQLAlchemy，正在为递归表Artifact实现CRUD方法，表结构如下：

class Artifact(Base):
    """
    Artifact table with defined columns and relationships.
    """

    __tablename__ = "artifact"

    id = db.Column(db.Integer, primary_key=True, nullable=False, autoincrement=True)
    parentId = db.Column(
        db.Integer, db.ForeignKey("artifact.id", name="artifact_parentId_fkey")
    )
    title = db.Column(db.String(100))
    notes = db.Column(db.String(500))
    startDate = db.Column(db.Date)
    endDate = db.Column(db.Date)
    extendedDate = db.Column(db.Date)
    artifactType = db.Column(db.Enum(ArtifactType))
    artifactStatus = db.Column(db.Enum(ArtifactStatus))
    artifactGroup = db.Column(db.Enum(ArtifactGroup))
    hourlyRate = db.Column(db.Numeric)
    price = db.Column(db.Numeric)
    taxes = db.Column(db.Numeric)
    effort = db.Column(db.Integer)
    settle = db.Column(db.Enum(SettleEnum))
    createInvoiceFor = db.Column(
        db.Date, default=lambda context: default_data(context, "startDate")
    )
    logo = db.Column(db.TEXT)
    contacts = db.Column(ARRAY(db.Integer))

    customerId = db.Column(db.Integer, db.ForeignKey("customer.id"))
    accesses = relationship("ResourceAccess", backref="artifact")

现有查询方法用于根据传入的根对象ID（通过filter_by传入），获取该对象所有嵌套父级、子级记录，同时通过ResourceAccess校验用户权限：

@classmethod
    def get_all_child_and_parent_records(
            cls, session, user_id: str, **filter_by
    ) -> List["Artifact"]:
        """
        Method to query all Artifact child and parent records from database.
        :param session: session maker
        :param user_id: accessing user ID.
        :param filter_by: Query filters
        :return: List[Artifact]
        """
        access_check = ResourceAccess.data_access(session, user_id)

        main_recursive = (
            session.query(Artifact)
            .filter_by(**{k: v for k, v in filter_by.items() if v is not None})
            .join(access_check, cls.id == access_check.c.artifactId)
            .cte("recursive", recursive=True)
        )

        child_query = (
            session.query(Artifact)
            .filter(Artifact.parentId == main_recursive.c.id)
            .join(access_check, cls.id == access_check.c.artifactId)
        )
        main_and_child_query = session.query(main_recursive.union_all(child_query))

        parent_query = (
            session.query(Artifact)
            .filter(Artifact.id == main_recursive.c.parentId)
            .join(access_check, cls.id == access_check.c.artifactId)
        )
        main_and_parent_query = session.query(main_recursive.union_all(parent_query))

        return main_and_child_query.all() + main_and_parent_query.all()[1:]

当前实现存在的问题：

• 重复查询数据库，需手动去除重复的根对象
• 尝试合并子查询和Union时，常出现父级/子级数据缺失、仅返回单层关联记录的情况

核心需求：

• 提升查询速度，简化代码
• 传入根对象ID后，返回所有嵌套父级、子级的Artifact记录
• 仅返回用户有权访问的记录（通过ResourceAccess校验）

功能示例：

• 查询Artifact 2，应返回父级Artifact1、子级Artifact4和Artifact5，不返回Artifact6和Artifact3
• 查询Artifact 4，应返回父级Artifact2、Artifact1，无返回子级，不返回Artifact5、Artifact6和Artifact3

通过单个递归CTE同时遍历父级和子级，避免两次查询和结果合并，同时自动去重，提升效率。

优化后代码 #

from sqlalchemy import union_all
from typing import List

@classmethod
def get_all_related_records(cls, session, user_id: str, **filter_by) -> List["Artifact"]:
    """
    获取指定Artifact的所有嵌套父级、子级记录，仅返回用户有权访问的条目
    :param session: 数据库会话
    :param user_id: 用户ID
    :param filter_by: 查询过滤条件（需包含根对象ID）
    :return: 相关Artifact记录列表
    """
    access_check = ResourceAccess.data_access(session, user_id)
    
    # 初始化递归CTE：先获取符合条件的根节点（已做权限校验）
    recursive_cte = (
        session.query(cls)
        .filter_by(**{k: v for k, v in filter_by.items() if v is not None})
        .join(access_check, cls.id == access_check.c.artifactId)
        .cte(name="artifact_recursive", recursive=True)
    )
    
    # 递归分支：同时查询当前节点的子节点和父节点，均做权限校验
    recursive_branch = union_all(
        # 查询子节点：匹配CTE中记录的id作为parentId
        session.query(cls)
        .filter(cls.parentId == recursive_cte.c.id)
        .join(access_check, cls.id == access_check.c.artifactId),
        # 查询父节点：匹配CTE中记录的parentId作为id
        session.query(cls)
        .filter(cls.id == recursive_cte.c.parentId)
        .join(access_check, cls.id == access_check.c.artifactId)
    )
    
    # 将递归分支合并到CTE中
    recursive_cte = recursive_cte.union_all(recursive_branch)
    
    # 查询去重后的最终结果
    return session.query(cls).from_statement(
        session.query(recursive_cte).distinct()
    ).all()

优化说明 #

1. 单CTE双向遍历：在同一个递归CTE中同时处理子节点和父节点查询，减少一次数据库交互，提升查询效率
2. 自动去重：通过distinct()确保根节点只出现一次，无需手动处理结果列表的重复项
3. 权限校验全程覆盖：所有递归查询分支都关联access_check，保证返回的每一条记录都经过权限校验
4. 代码简化：合并原有的两次查询逻辑，结构更清晰，降低维护成本

关键逻辑说明 #

• 递归CTE初始部分先获取符合过滤条件的根节点，同时完成权限校验
• 递归分支通过union_all同时拉取子节点（cls.parentId == recursive_cte.c.id）和父节点（cls.id == recursive_cte.c.parentId）
• 最终通过distinct()去重，确保结果无重复记录

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