如何用SQLAlchemy Core在PostgreSQL自增列批量插入并优化

一、替代实现方式与优化思路 #

你当前用循环单条插入的方式性能损耗极大，核心优化方向是减少数据库交互次数，结合PostgreSQL的SERIAL自增特性，以下是几种高效实现：

1. SQLAlchemy Core原生批量插入 #

利用insert() API批量构造插入请求，只需要一次数据库交互：

from sqlalchemy import insert, table, column

# 定义表映射对象（若已有可跳过）
id_gen_table = table(
    "id_generation",
    column("id")
)

# 构造批量插入的空值列表（SERIAL会自动填充自增ID）
values_list = [{} for _ in range(number_of_ids)]
conn.execute(insert(id_gen_table).values(values_list))
conn.commit()

该方式会生成一条批量插入SQL，将所有请求合并为一次数据库调用，性能远高于循环单条插入。

2. 借助PostgreSQL内置函数generate_series #

直接在数据库端生成所需行数，彻底避免Python与数据库间的数据传输，性能最优：

from sqlalchemy import text

conn.execute(text(f"""
    INSERT INTO id_generation 
    SELECT DEFAULT FROM generate_series(1, {number_of_ids})
"""))
conn.commit()

generate_series会生成指定范围的序列，配合SELECT DEFAULT直接触发SERIAL的自增逻辑，适合超大批量插入场景。

3. 分批次插入（超大数据量场景） #

如果number_of_ids达到百万级，一次性插入可能占用过多数据库资源，可按固定批次拆分：

batch_size = 1000
total_batches = number_of_ids // batch_size

for _ in range(total_batches):
    values_list = [{} for _ in range(batch_size)]
    conn.execute(insert(id_gen_table).values(values_list))
conn.commit()

或结合generate_series分批次：

batch_size = 1000
total_batches = number_of_ids // batch_size

for i in range(total_batches):
    start = i * batch_size + 1
    end = (i + 1) * batch_size
    conn.execute(text(f"""
        INSERT INTO id_generation 
        SELECT DEFAULT FROM generate_series({start}, {end})
    """))
conn.commit()

二、性能对比 #

• 循环单条插入：每次循环都要建立数据库连接、发送SQL、等待响应，IO瓶颈严重，性能最差。
• SQLAlchemy批量插入：合并请求为一次交互，性能提升明显，适合中小批量场景。
• PostgreSQL generate_series：数据库端原生处理，无数据传输开销，性能最优，适合大批量场景。

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