如何使用asyncpg将Python列表项批量插入Postgres并实现冲突增量更新

核心思路是借助PostgreSQL原生的INSERT ... ON CONFLICT（Upsert）语法，结合unnest函数直接处理传入的整数数组，全程只需要发起一次数据库请求即可完成2万条数据的写入或更新，性能远高于逐行插入。

前置要求 #

你存储整数的字段必须设置唯一约束（UNIQUE），否则无法触发冲突判断逻辑。以下示例假设表名为integer_records，存储整数的字段名为int_val，累加字段名为amount，可根据实际场景替换对应名称。

基础实现（直接传入原始整数列表） #

无需提前处理数据，直接把整个整数列表作为参数传入即可，列表中重复的整数会自动触发多次累加：

import asyncpg
import asyncio

async def batch_upsert_ints(int_list: list[int]):
    # 替换为实际数据库连接参数
    conn = await asyncpg.connect(
        user="数据库用户名",
        password="数据库密码",
        database="数据库名",
        host="127.0.0.1"
    )
    
    try:
        # 单条SQL完成全量操作
        await conn.execute(
            """
            INSERT INTO integer_records (int_val, amount)
            SELECT unnest($1::int[]), 1
            ON CONFLICT (int_val)
            DO UPDATE SET amount = integer_records.amount + 1
            """,
            int_list
        )
    finally:
        await conn.close()

# 调用示例
if __name__ == "__main__":
    # 模拟2万条整数数据
    test_list = [i for i in range(20000)]
    asyncio.run(batch_upsert_ints(test_list))

优化实现（提前统计重复值，性能更高） #

如果你的整数列表中存在大量重复值，可以先在Python侧统计每个整数的出现次数，再传入数据库，减少数据库侧的处理量：

import asyncpg
import asyncio
from collections import Counter

async def optimized_batch_upsert(int_list: list[int]):
    # 统计每个整数出现的次数
    int_counter = Counter(int_list)
    int_vals = list(int_counter.keys())
    count_vals = list(int_counter.values())

    conn = await asyncpg.connect(
        user="数据库用户名",
        password="数据库密码",
        database="数据库名",
        host="127.0.0.1"
    )
    
    try:
        await conn.execute(
            """
            INSERT INTO integer_records (int_val, amount)
            SELECT * FROM unnest($1::int[], $2::int[])
            ON CONFLICT (int_val)
            DO UPDATE SET amount = integer_records.amount + excluded.amount
            """,
            int_vals,
            count_vals
        )
    finally:
        await conn.close()

说明 #

• 两种实现都只发起一次数据库请求，2万条数据的处理耗时通常在毫秒级。
• 所有参数都通过asyncpg的预编译机制传入，不存在SQL注入风险。

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