如何将MySQL递归CTE查询转换为SQLAlchemy实现？

假设你已经通过SQLAlchemy定义了对应board_tbl的模型类，示例如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Board(Base):
    __tablename__ = 'board_tbl'
    board_rid = Column(Integer, primary_key=True)
    subject = Column(String)
    lv = Column(Integer)
    parent_rid = Column(Integer)

接下来可以用SQLAlchemy构建与原SQL逻辑完全一致的递归CTE查询：

from sqlalchemy import select, cast, String, func
from sqlalchemy.orm import Session

# 定义递归CTE的锚点查询（初始数据集）
cte = (
    select(
        Board.board_rid,
        Board.subject,
        Board.lv,
        Board.parent_rid,
        cast(Board.board_rid, String(100)).label('depth')
    )
    .where(Board.parent_rid == 2)
    .cte(recursive=True)  # 声明这是递归CTE，对应MySQL的WITH RECURSIVE
)

# 定义递归关联部分，关联CTE自身
recursive_part = (
    select(
        Board.board_rid,
        Board.subject,
        Board.lv,
        Board.parent_rid,
        func.concat(cte.c.depth, ",", Board.board_rid).label('depth')
    )
    .join(cte, Board.parent_rid == cte.c.board_rid)
)

# 合并锚点查询与递归部分
cte = cte.union_all(recursive_part)

# 最终查询并排序
final_query = (
    select(cte.c.board_rid, cte.c.subject, cte.c.parent_rid, cte.c.lv, cte.c.depth)
    .order_by(cte.c.depth)
)

# 执行查询（需提前创建好Session实例session）
results = session.execute(final_query).fetchall()

关键对应说明： #

• .cte(recursive=True)直接对应MySQL的WITH RECURSIVE语法
• cast(Board.board_rid, String(100))等价于原SQL的CAST(board_rid as CHAR(100))
• func.concat(cte.c.depth, ",", Board.board_rid)对应原SQL的CONCAT(c.depth, ",", b.board_rid)
• 用union_all合并锚点数据集和递归关联结果，保持原SQL的递归逻辑

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