使用Asyncpg的executemany批量插入JSON数据报错求助

错误原因 #

你遇到的invalid input in executemany() argument sequence element #0: expected a sequence, got dict错误，核心原因是executemany的第二个参数需要是序列的序列（比如元组或列表组成的列表），但你传入的是字典列表；而转字符串后报错是因为PostgreSQL的JSON/JSONB列接受Python原生字典，转成字符串会导致类型不兼容。

正确实现代码 #

1. 基础批量插入示例 #

直接用元组列表作为executemany的参数，不需要手动转JSON字符串：

import asyncio
import asyncpg

async def batch_insert_clusters():
    # 连接数据库
    conn = await asyncpg.connect(user='your_user', password='your_pass', database='your_db', host='127.0.0.1')
    
    try:
        # 待插入的数据：每个元素是(id, cluster_json字典)的元组
        cluster_data = [
            (1, {"nodes": [1,2,3], "label": "clusterA"}),
            (2, {"nodes": [4,5], "label": "clusterB"}),
            (3, {"nodes": [6,7,8,9], "label": "clusterC"})
        ]
        
        # 执行批量插入，$1、$2对应元组中的两个元素
        await conn.executemany(
            """INSERT INTO cluster (id, cluster_json) VALUES ($1, $2)""",
            cluster_data
        )
        await conn.commit()
    finally:
        await conn.close()

asyncio.run(batch_insert_clusters())

2. 处理原始字典列表 #

如果你的原始数据是字典列表（比如[{"id":1, "cluster_json": {...}}, ...]），先把每个字典转换成元组：

# 原始字典列表
raw_data = [
    {"id": 1, "cluster_json": {"nodes": [1,2,3], "label": "clusterA"}},
    {"id": 2, "cluster_json": {"nodes": [4,5], "label": "clusterB"}}
]

# 转换为元组列表
cluster_data = [(item['id'], item['cluster_json']) for item in raw_data]

# 后续executemany调用和基础示例一致

3. 关键注意事项 #

• 无需手动转JSON字符串：asyncpg会自动将Python字典序列化为PostgreSQL的JSON/JSONB类型，也能自动反向解析。
• 确保表的cluster_json列类型为JSON或JSONB：如果是TEXT类型才需要转字符串，但不推荐这种做法（会失去JSON类型的查询优势）。
• 超大数据集分批次插入：如果数据量过万，建议拆分批次插入，避免内存占用过高。

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