SQLAlchemy Asyncio批量插入多列失败，求高效实现方法

你的问题出在构造插入数据的方式上——你把每条记录拆成了两个字典的元组，而SQLAlchemy的insert()方法期望每个待插入的记录是单个包含所有字段键值对的字典。

正确的批量插入写法 #

直接构造一个包含完整记录的字典列表，一次性传给insert()执行，就能完成多列的批量插入，完全不需要后续的循环更新：

修改后的async_main()核心代码如下：

async def async_main() -> None:
    engine = create_async_engine(DB_URL, echo=True)

    async with engine.begin() as conn:
        await conn.run_sync(meta.drop_all)
        await conn.run_sync(meta.create_all)
        # 构造包含所有字段的完整记录列表
        insert_data = [
            {
                "first_name": f"first_name_{x:06d}",
                "last_name": f"last_name_{x:06d}",
                "email": f"email{x:06d}@nomail.io",
                "username": f"some name {x:06d}"
            }
            for x in range(NUM_RECORDS)
        ]
        # 单次执行批量插入
        await conn.execute(users.insert(), insert_data)

    await engine.dispose()

原写法效率低的原因 #

你之前的代码先插入first_name，再循环执行3次更新/每条记录，等于总共执行了1 + 3*NUM_RECORDS次SQL请求。而批量插入只需要1次请求，数据库只需要处理一次事务，性能提升非常明显，尤其是当NUM_RECORDS很大的时候。

错误写法的问题 #

你尝试的[({"first_name": ...}, {"last_name": ...}) for x in ...]把每条记录变成了字典元组，SQLAlchemy无法解析这种格式，因此抛出了类型错误。必须保证列表里的每个元素是单个完整的记录字典。

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