python - 在 SQLAlchemy 中处理插入时重复的主键(声明式样式)

Question

我的应用程序正在使用范围 session 和 SQLALchemy 的声明式样式。它是一个网络应用程序，许多数据库插入由任务调度程序 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 的人可能会工作到一半，而另一个正在检查它的存在。在这种情况下，第二个工作人员将尝试插入具有相同主键的行，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 acool thing 那么我想我可能会错过 IntegrityError 的东西，而这实际上并不是因为重复的主键。

有没有更好的方法，可以保持我正在使用的干净的 SQLAlchemy 方法(而不是开始在字符串中写出语句等......)

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

干杯!

Answer 1

如果您使用 session.merge(bike) 而不是 session.add(bike)，则不会产生主键错误。 bike 将根据需要进行检索和更新或创建。