<p>我的应用程序正在使用范围会话和SQLALchemy的声明式风格。它是一个web应用程序,许多DB插入都是由任务调度程序<code>Celery</code>执行的。</p>
<p>通常,当决定插入一个对象时,我的代码可能会沿着以下几行执行操作:</p>
<pre><code>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()
</code></pre>
<p>这里的问题是,因为很多工作都是由异步工作人员完成的,所以一个工作人员可能在插入带有<code>id=123</code>的<code>Bike</code>的过程中完成了一半,而另一个工作人员正在检查它的存在性。在这种情况下,第二个worker将尝试插入具有相同主键的行,SQLAlchemy将引发一个<code>IntegrityError</code>。</p>
<p>我一辈子都找不到解决这个问题的好办法,除了把<code>Session.commit()</code>换成:</p>
<pre><code>'''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()
</code></pre>
<p>然后在任何地方我都有<code>Session.commit</code>我现在有<code>schema.commit(ignore=True)</code>我不介意行不再被插入。</p>
<p>在我看来,这似乎很脆弱,因为字符串检查。正如仅供参考,当<code>IntegrityError</code>升高时,它看起来是这样的:</p>
<pre><code>(IntegrityError) (1062, "Duplicate entry '123' for key 'PRIMARY'")
</code></pre>
<p>当然,我插入的主键是<code>Duplicate entry is a cool thing</code>,我想我可能会漏掉<code>IntegrityError</code>,这并不是因为主键重复。</p>
<p>有没有更好的方法可以保持我使用的干净的SQLAlchemy方法(而不是开始用字符串写出语句等等…)</p>
<p>Db是MySQL(尽管对于单元测试,我喜欢使用SQLite,并且不想用任何新方法阻碍这种能力)。</p>
<p>干杯!</p>