使用 SQLAlchemy 使用子查询进行批量更新答案

【问题标题】：Bulk update with subquery using SQLAlchemy使用 SQLAlchemy 使用子查询进行批量更新
【发布时间】：2013-12-07 15:01:35
【问题描述】：

我正在尝试使用 SQLAlchemy 实现以下 MySQL 查询。有问题的表是嵌套集层次结构。

UPDATE category
JOIN
    (
    SELECT
        node.cat_id,
        (COUNT(parent.cat_id) - 1) AS depth
    FROM category AS node, category AS parent
    WHERE node.lft BETWEEN parent.lft AND parent.rgt
    GROUP BY node.cat_id
    ) AS depths
ON category.cat_id = depths.cat_id
SET category.depth = depths.depth

这很好用。

这是我开始拔头发的地方：

from sqlalchemy.orm import aliased
from sqlalchemy import func

from myapp.db import db

node = aliased(Category)
parent = aliased(Category)
stmt = db.session.query(node.cat_id,
                        func.count(parent.cat_id).label('depth_'))\
    .filter(node.lft.between(parent.lft, parent.rgt))\
    .group_by(node.cat_id).subquery()

db.session.query(Category,
                 stmt.c.cat_id,
                 stmt.c.depth_)\
    .outerjoin(stmt, 
               Category.cat_id == stmt.c.cat_id)\
    .update({Category.depth: stmt.c.depth_},
             synchronize_session='fetch')

...我得到InvalidRequestError: This operation requires only one Table or entity be specified as the target。在我看来，Category.depth 充分指定了目标，但当然 SQLAlchemy 胜过我的想法。

难住了。有什么建议？谢谢。

【问题讨论】：

你能在没有.subquery()的情况下打印出stmt吗？
你弄明白了吗？

标签： python mysql sql sqlalchemy nested-set-model

【解决方案1】：

我知道这个问题已经有五年的历史了，但我今天偶然发现了它。我的回答可能对其他人有用。我知道我的解决方案并不完美，但我没有更好的方法来做到这一点。

我只需要将最后一行更改为：

db.session.query(Category)\
    .outerjoin(stmt, 
               Category.cat_id == stmt.c.cat_id)\
    .update({Category.depth: stmt.c.depth_},
             synchronize_session='fetch')

然后，您必须提交更改：

db.session.commit()

这会给出以下警告：

SAWarning：在 ORM 上评估非映射列表达式“...” 实例；这是一个已弃用的用例。请利用 ORM 计算的 UPDATE / DELETE 表达式中的实际映射列。
“更新/删除表达式。” % 子句

为了摆脱它，我使用了这篇文章中的解决方案：Turn off a warning in sqlalchemy

注意：由于某种原因，别名在 SQLAlchemy 更新语句中不起作用。

【讨论】：