在SQLAlchemy（声明式样式）中处理插入时的重复主键

Question

我的应用程序正在使用范围会话和SQLALchemy的声明式风格。它是一个web应用程序，许多DB插入都是由任务调度程序Celery执行的。

通常，当决定插入一个对象时，我的代码可能会沿着以下几行执行操作：

from schema import Session
from schema.models import Bike

pk = 123 # primary key
bike = Session.query(Bike).filter_by(bike_id=pk).first()
if not bike: # no bike in DB
    new_bike = Bike(pk, "shiny", "bike")
    Session.add(new_bike)
    Session.commit()

这里的问题是，因为很多工作都是由异步工作人员完成的，所以一个工作人员可能在插入带有id=123的Bike的过程中完成了一半，而另一个工作人员正在检查它的存在性。在这种情况下，第二个worker将尝试插入具有相同主键的行，SQLAlchemy将引发一个IntegrityError。

我一辈子都找不到解决这个问题的好办法，除了把Session.commit()换成：

'''schema/__init__.py'''
from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker())

def commit(ignore=False):
    try:
        Session.commit()
    except IntegrityError as e:
        reason = e.message
        logger.warning(reason)

        if not ignore:
            raise e

        if "Duplicate entry" in reason:
            logger.info("%s already in table." % e.params[0])
            Session.rollback()

然后在任何地方我都有Session.commit我现在有schema.commit(ignore=True)我不介意行不再被插入。

在我看来，这似乎很脆弱，因为字符串检查。正如仅供参考，当IntegrityError升高时，它看起来是这样的：

(IntegrityError) (1062, "Duplicate entry '123' for key 'PRIMARY'")

当然，我插入的主键是Duplicate entry is a cool thing，我想我可能会漏掉IntegrityError，这并不是因为主键重复。

有没有更好的方法可以保持我使用的干净的SQLAlchemy方法（而不是开始用字符串写出语句等等…）

Db是MySQL（尽管对于单元测试，我喜欢使用SQLite，并且不想用任何新方法阻碍这种能力）。

干杯！

Answer 1

您应该以同样的方式处理每一个IntegrityError：回滚事务，然后有选择地重试。有些数据库甚至不允许您在一个IntegrityError之后做更多的事情。您还可以在两个冲突事务开始时获取表上的锁，或者在数据库允许的情况下获取更细粒度的锁。

使用with语句显式开始事务，并自动提交（或回滚任何异常）：

from schema import Session
from schema.models import Bike

session = Session()
with session.begin():
    pk = 123 # primary key
    bike = session.query(Bike).filter_by(bike_id=pk).first()
    if not bike: # no bike in DB
        new_bike = Bike(pk, "shiny", "bike")
        session.add(new_bike)