SQLAlchemy简单递归cte查询

Question

我有以下 SQLAlchemy 表：

\n\n\n\n

from sqlalchemy.ext.declarative import declarative_base\n\nBase = declarative_base()\n\nclass NetworkLink(Base):\n    """Network immediate link between a franchisee and his franchisor\n\n    """\n    __tablename__ = \'network_link\'\n\n    id_franchisee = Column(Integer, ForeignKey(\'user.id\'), primary_key=True)\n    id_franchisor = Column(Integer, ForeignKey(\'user.id\'))\n

\n\n

它基本上代表了一个树状的网络结构。

\n\n

给定特许人的 id，我需要获取整个子树中所有后代的 id。\n例如，如果表如下：

\n\n

id_franchisor | id_franchisee \n1 | 2\n1 | 3\n2 | 4\n2 | 5\n4 | 6\n

\n\n

然后给定 id 1 我需要 1,2,3,4,5,6，而给定 2 我需要 2,4,5,6。

\n\n

我知道这不是解决此问题的最有效的表表示形式，但此操作将很少执行，并且插入将更加常见。

\n\n

我正在尝试使用递归查询来实现此功能，如下所示：

\n\n

"""\nWITH RECURSIVE recursive_franchisee(id) AS\n(\n    SELECT %s\n    UNION ALL\n    SELECT L.id_franchisee\n    FROM recursive_franchisee as R JOIN network_link as L ON R.id = L.id_franchisor\n) SELECT id FROM recursive_franchisee;\n"""\n

\n\n

我可以在哪里替换%s我想要的 id。我已经测试过它并且工作正常。\n但是我想避免使用硬编码的原始查询，因此我尝试在 SQLAlchemy 中重新创建它，但失败了。\n查看中的文档\n https://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.cte \n我想出了两个替代方案，足够接近但不太有效。

\n\n

第一个几乎是正确的，但我不知道如何指定起始 id：

\n\n

rec = db_session.query("id").cte(recursive=True, name="recursive_franchisee")\nralias = sqlalchemy.orm.aliased(rec, name="R")\nlalias = sqlalchemy.orm.aliased(NetworkLink, name="L")\nrec = rec.union_all(\n    db_session.query(lalias.id_franchisee) \\\n              .join(ralias, ralias.c.id == lalias.id_franchisor)\n)\nqr = db_session.query(rec)\nsqr = str(qr)\n\n# sqr equals to:\n"""\nWITH RECURSIVE recursive_franchisee(id) AS\n(\n    SELECT id  # how do I specify an id here?\n    UNION ALL\n    SELECT "L".id_franchisee AS "L_id_franchisee"\n    FROM network_link AS "L" JOIN recursive_franchisee AS "R" ON "R".id = "L".id_franchisor\n)\nSELECT recursive_franchisee.id FROM recursive_franchisee\n"""\n

\n\n

我还尝试了以下方法，但有不同的问题：

\n\n

rec = db_session.query(NetworkLink.id_franchisor) \\\n                .filter(NetworkLink.id_franchisor == 1) \\\n                .cte(recursive=True, name="recursive_franchisee")\nralias = sqlalchemy.orm.aliased(rec, name="R")\nlalias = sqlalchemy.orm.aliased(NetworkLink, name="L")\nrec = rec.union_all(\n    db_session.query(lalias.id_franchisee) \\\n              .join(ralias, ralias.c.id_franchisor == lalias.id_franchisor)\n)\nqr = db_session.query(rec)\nsqr = str(qr)\n\n# sqr equals to:\n"""\nWITH RECURSIVE recursive_franchisee(id_franchisor) AS\n(\n    SELECT network_link.id_franchisor AS id_franchisor\n    FROM network_link\n    WHERE network_link.id_franchisor = ?\n    UNION ALL\n    SELECT "L".id_franchisee AS "L_id_franchisee"\n    FROM network_link AS "L" JOIN recursive_franchisee AS "R" ON "R".id_franchisor = "L".id_franchisor\n)\nSELECT recursive_franchisee.id_franchisor AS recursive_franchisee_id_franchisor\nFROM recursive_franchisee\n"""\n

\n\n

这当然会产生重复的结果，因为第一个结果SELECT会多次返回特许人（每个特许经营者一个）。我想我可以使用DISTINCT，但我想避免选择我一开始就知道的东西。

\n\n

如何在 SQLAlchemy 中实现我需要的查询？

\n\n

编辑：

\n\n

根据 Ilja Everil\xc3\xa4\ 的评论，我能够提出以下工作解决方案：

\n\n

from sqlalchemy.sql.expression import literal\n\nrec = db_session.query(literal(current_user.id).label("id")) \\\n                .cte(recursive=True, name="recursive_franchisee")\nralias = sqlalchemy.orm.aliased(rec, name="R")\nlalias = sqlalchemy.orm.aliased(NetworkLink, name="L")\nrec = rec.union_all(\n    db_session.query(lalias.id_franchisee) \\\n            .join(ralias, ralias.c.id == lalias.id_franchisor)\n)\nqr = db_session.query(rec)\n

\n\n

谢谢您，如果您发布答案，我会接受。

\n