如何使用SQLAlchemy Connection.execute()向INSERT传入多组参数？

以下是PostgreSQL的有效SQL语句：

INSERT INTO schema.table (id, letter)
VALUES 
    (1, 'a'),
    (2, 'b'),
    ...

我希望使用SQLAlchemy执行类似的参数化语句，当前使用SQLAlchemy 1.4版本的Connection.execute()，但无法访问表映射类（ORM模型）。

以下代码无法正常运行，但能体现我的需求：

statement: str = """
    INSERT INTO schema.table (id, letter)
    VALUES :values
"""

values: Tuple[Tuple[int, str], ...] = (
    (1, "a"),
    (2, "b"),
)

with engine.connect() as connection:
    connection.execute(
        sqlalchemy.text(statement),
        {"values": values},
    )

在这个示例中，values是一个元组，它会被绑定为元组的元组，不符合需求（报错“INSERT has more expressions than target columns”）。同样，使用元组列表会生成SQL ARRAY。

问：为什么不直接使用f-string或"...".format(...)？
答：我了解这种做法属于不良实践。

问题： 如何正确“解包”values参数？或者，在不使用Table ORM类的情况下，实现该需求的最佳方式是什么？

注：我知晓提问时SQLAlchemy 1.4已处于弃用状态。

方法1：用executemany模式批量传入参数 #

这是最简洁的方式，SQLAlchemy会自动处理参数化和批量插入逻辑：

from sqlalchemy import text

# 定义单条插入的SQL模板
statement = text("""
    INSERT INTO schema.table (id, letter)
    VALUES (:id, :letter)
""")

# 以列表字典的形式组织多条数据
values = [
    {"id": 1, "letter": "a"},
    {"id": 2, "letter": "b"},
]

with engine.connect() as connection:
    connection.execute(statement, values)
    connection.commit()  # 手动管理事务时需要提交

方法2：用insert()构造器（无需ORM模型） #

直接用SQLAlchemy的表达式构造器生成插入语句，无需依赖映射类：

from sqlalchemy import insert, text

# 用text()指定带schema的表名，传入列表字典格式的数据
insert_stmt = insert(text("schema.table")).values(
    [{"id": 1, "letter": "a"}, {"id": 2, "letter": "b"}]
)

with engine.connect() as connection:
    connection.execute(insert_stmt)
    connection.commit()

这种方式符合SQLAlchemy的规范，自动处理参数绑定，避免手动拼接的风险。

方法3：动态生成占位符（手动参数展开） #

如果需要更灵活的占位符控制，可以动态生成对应数量的占位符，再扁平化参数传入：

from sqlalchemy import text

values = (
    (1, "a"),
    (2, "b"),
)

# 根据数据条数生成对应的VALUES占位符
placeholders = ", ".join([f"(:v{i*2+1}, :v{i*2+2})" for i in range(len(values))])
statement = text(f"""
    INSERT INTO schema.table (id, letter)
    VALUES {placeholders}
""")

# 把嵌套元组转成一维参数字典
params = {}
for idx, (id_val, letter_val) in enumerate(values):
    params[f"v{idx*2+1}"] = id_val
    params[f"v{idx*2+2}"] = letter_val

with engine.connect() as connection:
    connection.execute(statement, params)
    connection.commit()

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