Pyramid与SQLAlchemy中如何正确处理重复主键问题？

正确处理方案 #

在Pyramid和SQLAlchemy组合环境中，处理重复主键这类数据库完整性异常，核心是捕获异常后回滚未提交事务，同时清理会话脏数据以保留会话可用性，避免后续操作报错。

优化后的add_record实现

from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import FlushError

def add_record(request, code, name):
    try:
        new_record = MyModel(id=code, name=name, value=1)
        request.dbsession.add(new_record)
        request.dbsession.flush()
        return True, ""
    except (IntegrityError, FlushError) as e:
        # 回滚当前未提交的事务操作
        request.dbsession.rollback()
        # 清理会话中残留的未提交对象，消除脏数据影响
        request.dbsession.expunge_all()
        message = f"Code {code} already exists"
        print(message)
        return False, message

关键细节说明

• 捕获完整的异常类型：除了数据库层面的IntegrityError，还要处理SQLAlchemy会话级的FlushError，覆盖flush阶段可能出现的异常场景。
• rollback后清理会话：调用expunge_all()移除会话中所有未提交的对象，确保后续使用同一个session时不会出现脏数据冲突，避免触发PendingRollbackError。
• 不要手动关闭会话：Pyramid的请求级session会在请求生命周期结束后自动管理，手动关闭会导致后续操作抛出ResourceClosedError。

额外优化建议

可以在添加记录前先做一次主键存在性查询，减少异常触发的概率，但注意并发场景下仍可能出现竞争，所以异常捕获仍是必不可少的兜底方案：

def add_record(request, code, name):
    # 前置查询减少异常触发
    existing = request.dbsession.query(MyModel).filter_by(id=code).first()
    if existing:
        return False, f"Code {code} already exists"
    
    try:
        new_record = MyModel(id=code, name=name, value=1)
        request.dbsession.add(new_record)
        request.dbsession.flush()
        return True, ""
    except (IntegrityError, FlushError) as e:
        request.dbsession.rollback()
        request.dbsession.expunge_all()
        message = f"Code {code} already exists"
        print(message)
        return False, message

内容的提问来源于stack exchange，提问作者QLands - Carlos Quiros