如何规避asyncpg查询参数限制？超32767参数最优插入方案

问题原因 #

你遇到的the number of query arguments cannot exceed 32767错误，确实是PostgreSQL wire协议的限制——Bind消息的参数字段用int16存储，最大只能容纳32767个参数。而psycopg2之所以能处理大量数据，是因为它会自动优化批量插入语句：将多行VALUES打包成数组参数，再通过unnest函数展开，把参数数量从「行数×列数」降到「列数」，绕过了协议限制。asyncpg的SQLAlchemy适配器默认没有这个优化，所以直接传入大量行数据会触发报错。

无需分批的最优解决方案 #

方案1：手动模拟psycopg2的数组+unnest优化 #

通过SQLAlchemy的from_select配合unnest，将多行数据按列打包成数组，把参数数量压缩到列的数量，完全避开32767的限制。

示例代码：

from sqlalchemy import insert, select, func, bindparam
from sqlalchemy.dialects.postgresql import array
from your_module import your_table  # 导入你的目标表模型

# 假设data是包含40万条记录的列表，每个元素是dict（键对应列名）
column_names = ["column1", "column2", "column3", "column4", "column5"]

# 按列提取数据，生成PostgreSQL数组参数
column_arrays = {
    f"{col}_array": array([row[col] for row in data])
    for col in column_names
}

# 构建插入语句：从unnest的结果中选择数据插入
stmt = insert(your_table).from_select(
    column_names,
    select(
        func.unnest(bindparam("column1_array")).label("column1"),
        func.unnest(bindparam("column2_array")).label("column2"),
        func.unnest(bindparam("column3_array")).label("column3"),
        func.unnest(bindparam("column4_array")).label("column4"),
        func.unnest(bindparam("column5_array")).label("column5"),
    ).params(**column_arrays)
).on_conflict_do_update(
    index_elements=["column1"],
    set_={col: insert(your_table).excluded[col] for col in column_names[1:]}
)

# 异步执行语句
async with async_session() as session:
    await session.execute(stmt)
    await session.commit()

方案2：使用COPY命令+临时表（超大量数据首选） #

COPY是PostgreSQL效率最高的批量导入方式，完全不受参数数量限制。如果需要处理冲突更新，可以先将数据COPY到临时表，再通过INSERT...ON CONFLICT从临时表同步到目标表。

示例代码：

import io
from sqlalchemy.ext.asyncio import AsyncSession
from your_module import your_table  # 导入你的目标表模型

async def bulk_insert_with_conflict(session: AsyncSession, data):
    # 创建与目标表结构一致的临时表
    temp_table = your_table.__table__.temporary_table()
    await session.run_sync(lambda sync_db: sync_db.execute(temp_table.create()))

    # 将数据写入内存CSV缓冲区
    buffer = io.StringIO()
    for row in data:
        # 按列顺序拼接（注意处理特殊字符，比如换行、制表符）
        row_values = [str(row[col]) for col in your_table.columns.keys()]
        buffer.write("\t".join(row_values) + "\n")
    buffer.seek(0)

    # 获取asyncpg底层连接，执行COPY导入临时表
    conn = await session.connection()
    async with conn.transaction():
        await conn.connection.cursor.copy_from(
            buffer,
            temp_table.name,
            columns=temp_table.columns.keys()
        )

        # 从临时表插入目标表，处理冲突更新
        stmt = insert(your_table).from_select(
            your_table.columns.keys(),
            select(temp_table.columns)
        ).on_conflict_do_update(
            index_elements=["column1"],
            set_={col: insert(your_table).excluded[col] for col in your_table.columns if col != "column1"}
        )
        await session.execute(stmt)

        # 删除临时表
        await session.run_sync(lambda sync_db: sync_db.execute(temp_table.drop()))

# 调用示例
async with async_session() as session:
    await bulk_insert_with_conflict(session, your_400k_rows_data)
    await session.commit()

为什么psycopg2没有这个问题？ #

psycopg2在处理executemany或批量VALUES插入时，会自动做「数组打包+unnest展开」的优化。比如你传入10000行×5列的数据，psycopg2会把它转换成5个数组参数（每列一个数组），然后生成INSERT ... SELECT unnest($1), unnest($2), ...的语句，参数数量只有5个，远低于32767的限制。而asyncpg的SQLAlchemy适配器目前没有内置这个优化逻辑，所以需要手动处理。

内容的提问来源于stack exchange，提问作者Sergey Bakaev Rettley