如何将该PostgreSQL SQL查询转换为SQLAlchemy Core？

直接上对应实现代码，搭配注释说明：

首先定义表结构（如果还没定义的话）：

from sqlalchemy import MetaData, Table, Column, String, Integer
from sqlalchemy.sql import update, values

metadata = MetaData()
test = Table(
    'test', metadata,
    Column('column_b', String, primary_key=True),
    Column('column_a', Integer)
)

然后构建并执行目标UPDATE语句：

# 准备要批量更新的数据
batch_data = [('123', 1), ('345', 2)]

# 构造PostgreSQL VALUES子句，指定列名和对应类型
values_clause = values(batch_data).columns(column_b=String, column_a=Integer)

# 构建核心UPDATE语句，对应你给出的原生SQL
update_stmt = (
    update(test)
    .values(column_a=values_clause.c.column_a)
    .from_(values_clause)
    .where(values_clause.c.column_b == test.c.column_b)
)

# 执行语句（假设你已有数据库连接对象connection）
# connection.execute(update_stmt)

关键说明 #

• sqlalchemy.sql.values()用来生成原生SQL中的VALUES (...) AS c(...)部分，通过.columns()显式指定列名与类型，确保和原表字段匹配
• update()对象的.from_()方法对应PostgreSQL特有的UPDATE ... FROM语法
• 通过where子句建立临时VALUES表和目标表的关联条件，和原生SQL逻辑完全一致

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