如何在SQLAlchemy ORM中实现带条件的批量更新？

需求 #

批量更新MyTable中ID在LIST_OF_IDS列表里的记录：

• 当old_value不为None时，将new_value设为old_value的值
• 当old_value为None时，将new_value设为字符串'default_value'

模型类代码 #

class MyTable(BaseModel):
    id = sa.Column(sa.String(256))
    new_value = sa.Column(sa.String(256))
    old_value = sa.Column(sa.String(256))

尝试的错误写法 #

LIST_OF_IDS = [1,2,5,6]

update_query = db.session.query(MyTable).filter(MyTable.id.in_(LIST_OF_IDS)) \
     .update(dict(new_value=MyTable.old_value if MyTable.old_value else 'default_value'))

正确解决方案 #

你需要用SQLAlchemy的case构造实现条件逻辑，Python层面的if判断会在代码执行阶段求值，无法生成对应的SQL条件语句。

正确的更新代码如下：

from sqlalchemy import case

LIST_OF_IDS = [1,2,5,6]

update_query = db.session.query(MyTable).filter(MyTable.id.in_(LIST_OF_IDS)) \
    .update(
        {
            MyTable.new_value: case(
                (MyTable.old_value.isnot(None), MyTable.old_value),
                else_='default_value'
            )
        },
        synchronize_session=False  # 批量更新时通常设为False提升性能，需同步内存对象可设为'fetch'
    )

db.session.commit()  # 必须提交事务完成更新

说明 #

• case语句会生成SQL的CASE表达式，在数据库层面完成条件判断
• synchronize_session参数：无需同步内存对象时设为False，需要同步则设为'fetch'

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