在 SQLAlchemy 中插入或更新重复键（IntegrityError）答案

【问题标题】：Insert or update duplicate key in SQLAlchemy (IntegrityError)在 SQLAlchemy 中插入或更新重复键（IntegrityError）
【发布时间】：2014-06-18 16:37:58
【问题描述】：

我正在尝试解析一些 RSS 提要并使用 flask-sqlalchemy 将 URL 写入数据库。

其中一些提要重叠（即同一篇文章出现在多个提要中或多次出现在同一个提要中），因此我将主键定义为 URL 的（非随机）散列。

我的问题是，当我遍历 URL 并将它们添加到 Session 时，当我尝试将它们提交到数据库时遇到异常。

这是我的代码：

for entry in feed.entries:
    # Create the database row
    article = Article(entry)

    # If the row already exists in the database
    if db.session.query(Article).filter_by(uuid=article.uuid).first():
        print "duplicate"
    else:
        db.session.merge(article)

db.session.commit()

当文章已经存在于数据库中时，它会被忽略。但是，如果它存在于 Session 中但尚未提交数据库，那么 SQLAlchemy 会尝试在同一个事务中多次写入它，我得到sqlalchemy.exc.IntegrityError: (IntegrityError) column hash is not unique。

我的直觉是我需要检查会话中是否不存在具有该哈希的对象（以及查询数据库），我认为这就是合并会做的，但我仍然得到错误。

【问题讨论】：

你试过了吗：if article in db.session or db.session.query(Article).filter_by(uuid=article.uuid).first():？（刚刚在文档中找到它：docs.sqlalchemy.org/en/rel_0_9/orm/… 它可能根本不起作用）

标签： python flask sqlalchemy

【解决方案1】：

您应该使用Unique Object pattern from the examples。

这会为会话创建一个内存缓存，并保证一个唯一的结果。如果实例在缓存中，则使用它，否则尝试从数据库中获取它。如果它不在数据库中，请创建它。将实例添加到缓存中。

这是我对该模式的看法，它简化了链接示例。

class UniqueMixin(object):
    @classmethod
    def get_unique(cls, **kwargs):
        session = current_app.extensions['sqlalchemy'].db.session
        session._unique_cache = cache = getattr(session, '_unique_cache', {})
        key = (cls, tuple(kwargs.items()))
        o = cache.get(key)

        if o is None:
            o = session.query(cls).filter_by(**kwargs).first()

            if o is None:
                o = cls(**kwargs)
                session.add(o)

            cache[key] = o

        return o

class MyModel(UniqueMixin, db.Model):
    # ...
    name = db.Column(db.String, nullable=False)
    # ...

m1 = MyModel.get_unique(name='test')  # new instance
m2 = MyModel.get_unique(name='test')  # from cache
assert m1 is m2
db.session.commit()  # inserts one row

m1 = MyModel.get_unique(name='test')  # from database
m2 = MyModel.get_unique(name='test')  # from cache
assert m1 is m2

在任何可能遇到这种情况的模型中继承 UniqueMixin，并使用 get_unique 代替普通的构造函数。

【讨论】：