多线程Python应用SQLAlchemy批量插入PostgreSQL的最优方案及冲突处理

你的问题核心是PostgreSQL序列值与表中实际最大ID不同步，高并发场景下SQLAlchemy预生成的ID已被其他会话插入，触发UniqueViolation异常。常见诱因：

1. 曾手动插入过指定ID的行，未同步更新序列；
2. 高并发下多会话同时请求序列值，SQLAlchemy会话缓存的ID与数据库实际状态冲突。

1. 异常捕获与重试机制 #

通过捕获IntegrityError，回滚会话后重试插入。重试时SQLAlchemy会重新从数据库获取序列值，避开重复ID。

修改你的predict函数：

from sqlalchemy.exc import IntegrityError

def predict(image_id: int, image_key: str):
    result = detector.detect(image_key, classes)
    max_retries = 3  # 设定合理的重试次数
    retries = 0
    
    while retries < max_retries:
        db_session = db()
        try:
            for i in result:
                db_prediction = models.Prediction(
                    # 不要传入id，交由数据库自动分配
                    image_id=image_id,
                    x1=i["x1"],
                    y1=i["y1"],
                    x2=i["x2"],
                    y2=i["y2"],
                    class_id=i["class_id"],
                    confidence=i["confidence"]
                )
                db_session.add(db_prediction)
            db_session.commit()
            return
        except IntegrityError:
            retries += 1
            db_session.rollback()  # 回滚未提交的操作
            if retries >= max_retries:
                raise  # 超过重试次数后抛出异常
        finally:
            db_session.close()  # 确保会话关闭

2. 修复序列与表ID同步（根本解决） #

先手动将序列值更新为表中最大ID+1，从根源避免ID冲突：

-- 执行该SQL修复序列
SELECT setval('public.predictions_id_seq', (SELECT MAX(id) FROM public.predictions) + 1);

若后续需手动插入ID，操作后务必同步更新序列，或尽量避免手动指定自增ID。

3. 优化模型定义 #

你的id字段中unique=True是冗余配置——primary_key=True已隐含唯一约束，可移除：

class Prediction(Base):
    __tablename__ = "predictions"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    # 其他字段...

4. 批量插入优化（可选） #

高并发下批量插入可减少会话开销，同时让数据库统一分配ID，降低冲突概率：

from sqlalchemy.exc import IntegrityError

def predict(image_id: int, image_key: str):
    result = detector.detect(image_key, classes)
    max_retries = 3
    retries = 0
    
    while retries < max_retries:
        db_session = db()
        try:
            # 构造批量插入对象列表
            predictions = [
                models.Prediction(
                    image_id=image_id,
                    x1=i["x1"],
                    y1=i["y1"],
                    x2=i["x2"],
                    y2=i["y2"],
                    class_id=i["class_id"],
                    confidence=i["confidence"]
                ) for i in result
            ]
            db_session.bulk_save_objects(predictions)
            db_session.commit()
            return
        except IntegrityError:
            retries += 1
            db_session.rollback()
            if retries >= max_retries:
                raise
        finally:
            db_session.close()

• 重试是临时应对高并发冲突的方案，修复序列同步才是根本解决途径；
• 禁止随意手动指定自增ID，否则会破坏序列自动递增逻辑；
• 高并发场景下，确保每个请求使用独立数据库会话，避免会话共享引发的ID冲突。

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