如何在Python中使用单条INSERT INTO语句批量插入多行数据？

问题背景 #

我正在开发Discord Python机器人，当前遍历ForumTags列表，为每个标签生成单独的INSERT INTO语句插入MySQL。希望优化为单条批量插入语句，示例SQL如下：

INSERT INTO guild_support_tags (guild_id, tag_id, tag_name, category) VALUES 
(123, 1, "test", "test_category"), (456, 2, "another tag", "test_category2")

当前使用SQLAlchemy+aiomysql（Python3.12），现有代码：

start = time.time()
for tag in forum.available_tags:
    await write_query("INSERT INTO guild_support_tags (guild_id, tag_id, tag_name, category) VALUES "
                      "(:guild_id, :tag_id, :tag_name, :category)", 
                      {"guild_id": interaction.guild_id, "tag_id": tag.id, "tag_name": tag.name, 
                       "category": category})

    print(f"loop done after {time.time() - start}")


# 外部执行函数，优先不修改
async def write_query(query: str, params: dict) -> None:
    async with async_session() as session:
        async with session.begin():
            await session.execute(text(query), params)

解决方案 #

方案1：不修改write_query函数 #

通过构造带索引的参数占位符，拼接批量插入SQL，适配原有函数：

start = time.time()
if not forum.available_tags:
    print("No tags to insert")
    return

values_clauses = []
params = {}
# 遍历标签，生成带索引的参数和VALUES子句
for idx, tag in enumerate(forum.available_tags, 1):
    params[f"guild_id_{idx}"] = interaction.guild_id
    params[f"tag_id_{idx}"] = tag.id
    params[f"tag_name_{idx}"] = tag.name
    params[f"category_{idx}"] = category
    values_clauses.append(f"(:guild_id_{idx}, :tag_id_{idx}, :tag_name_{idx}, :category_{idx})")

# 拼接完整批量插入语句
query = f"INSERT INTO guild_support_tags (guild_id, tag_id, tag_name, category) VALUES {', '.join(values_clauses)}"

# 执行批量插入
await write_query(query, params)
print(f"Batch insert done after {time.time() - start}")

方案2：优化write_query支持批量参数（更简洁） #

如果允许修改底层执行函数，让其支持参数列表，代码会更简洁：

# 修改后的write_query，支持单字典或字典列表
async def write_query(query: str, params: dict | list[dict]) -> None:
    async with async_session() as session:
        async with session.begin():
            await session.execute(text(query), params)

对应的批量插入代码：

start = time.time()
if not forum.available_tags:
    print("No tags to insert")
    return

# 构造参数列表
params_list = [
    {
        "guild_id": interaction.guild_id,
        "tag_id": tag.id,
        "tag_name": tag.name,
        "category": category
    }
    for tag in forum.available_tags
]

# 执行批量插入
query = "INSERT INTO guild_support_tags (guild_id, tag_id, tag_name, category) VALUES (:guild_id, :tag_id, :tag_name, :category)"
await write_query(query, params_list)
print(f"Batch insert done after {time.time() - start}")

方案说明 #

• 方案1完全兼容原有write_query，无需改动底层逻辑，适合需要保持现有代码结构的场景。
• 方案2通过扩展write_query的参数支持，利用SQLAlchemy原生的批量处理能力，代码更简洁，性能表现更优。
• 两种方案均避免了循环执行单条INSERT的网络和数据库开销，标签数量越多，性能提升越明显。

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