通过Python向PostgreSQL 14批量导入数据时统一生成递增运行编号的最佳实践问询

你的这个场景确实是数据库批量插入中很常见的需求，先直接说结论：你当前的两次调用方案并不是最佳实践——如果有多个进程/线程同时执行“查询最大编号→加1→插入”的操作，它们大概率会拿到相同的最大编号，最终插入重复的运行编号，导致数据不一致。

下面给你梳理几种更安全、高效的方案，按推荐程度排序：

方案1：序列（Sequence）+ 单次SQL插入（高并发首选） #

这是最稳妥且高效的方案，完全规避竞态问题：

第一步：创建专用序列 #

先在PostgreSQL中创建一个序列，专门用来生成唯一的运行编号：

CREATE SEQUENCE run_number_seq START WITH 1 INCREMENT BY 1;

序列是PostgreSQL原生的原子性自增机制，每次调用nextval()都会得到全局唯一的数值，不会出现重复。

第二步：单次SQL完成插入 #

通过WITH子句一次性获取新的运行编号，同时插入所有数据，整个操作是原子性的：

WITH new_run_id AS (
    SELECT nextval('run_number_seq') AS run_id
)
INSERT INTO your_table (run_number, column1, column2)
SELECT run_id, data_col1, data_col2
FROM new_run_id
CROSS JOIN (
    -- 这里替换成你要插入的数据集，Python中可以用参数绑定传入
    VALUES 
        ('value1_1', 'value1_2'),
        ('value2_1', 'value2_2')
) AS temp_data(data_col1, data_col2);

在Python代码里，你只需要把待插入的参数绑定到这个SQL语句中，一次调用就能完成所有操作，既安全又高效。

方案2：WITH子句计算最大编号+单次插入（低并发场景适用） #

如果不想依赖序列，也可以用原子化的SQL语句在一次调用中完成“获取新编号+插入”：

WITH new_run_id AS (
    -- 用COALESCE处理表为空的情况，默认从1开始
    SELECT COALESCE(MAX(run_number), 0) + 1 AS run_id FROM your_table
)
INSERT INTO your_table (run_number, column1, column2)
SELECT run_id, data_col1, data_col2
FROM new_run_id
CROSS JOIN (
    VALUES 
        ('value1_1', 'value1_2'),
        ('value2_1', 'value2_2')
) AS temp_data(data_col1, data_col2);

⚠️ 注意：这个方案在高并发场景下有风险——PostgreSQL默认的READ COMMITTED隔离级别下，MAX(run_number)不会读取其他未提交事务的插入结果，如果两个事务同时执行，可能会生成相同的编号。解决办法：

• 把事务隔离级别提升为REPEATABLE READ
• 或者在SELECT MAX(...)后加上FOR UPDATE锁定表（但会降低并发性能）

所以这个方案更适合低并发、数据量不大的场景。

方案3：封装为自定义函数（逻辑复用首选） #

如果这个插入逻辑会被多个地方复用，推荐把它封装成PostgreSQL的自定义函数，这样Python端的代码会更简洁，逻辑也更集中：

创建函数 #

CREATE OR REPLACE FUNCTION insert_with_run_number(p_data JSONB)
RETURNS VOID AS $$
DECLARE
    new_run_id INT;
BEGIN
    -- 这里用序列生成编号，也可以换成MAX+1的逻辑
    new_run_id := nextval('run_number_seq');
    
    -- 解析JSON数组并插入数据
    INSERT INTO your_table (run_number, column1, column2)
    SELECT new_run_id, (elem->>'column1')::TEXT, (elem->>'column2')::TEXT
    FROM jsonb_array_elements(p_data) AS elem;
END;
$$ LANGUAGE plpgsql VOLATILE;

Python调用函数 #

在Python中，把待插入的数据整理成JSON数组，直接调用函数即可：

import psycopg2
import json

# 待插入的数据
batch_data = [
    {"column1": "value1", "column2": "value2"},
    {"column1": "value3", "column2": "value4"}
]

# 数据库连接与调用
conn = psycopg2.connect("your_connection_string")
cur = conn.cursor()
cur.execute("SELECT insert_with_run_number(%s)", (json.dumps(batch_data),))
conn.commit()
cur.close()
conn.close()

这种方式的好处是把数据库操作逻辑封装在数据库端，后续如果需要调整运行编号的生成规则，只需要修改函数，不用改动Python代码。

总结一下 #

1. 绝对避免两次调用的方案，竞态风险太高；
2. 高并发场景优先选序列+单次SQL的方案；
3. 低并发场景可以用MAX+1的单次SQL方案；
4. 如果逻辑复用率高，推荐封装为自定义函数。

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