python - 让 SQLAlchemy 在 ORM 级联中执行 "ON DUPLICATE KEY UPDATE"(在 MySQL 中)

Question

我从 python reddit channel 交叉发布这个，我还没有设法得到对此的回应。我很惊讶，因为这似乎是一个非常标准的场景。

我的问题是，在 SQLAlchemy 的 ORM 接口(interface)(在 mysql 上)中，是否有一种简单的方法来处理级联插入中子对象的主键冲突。举一个具体的例子，我有一个句子表和一个“字典”表，当我创建一个句子记录时，我想在字典中插入所有相关的单词。当然，很多时候有些词已经在字典里了，所以我得到一个主键错误(这个词是关键)。如果我使用原始 SQL，我会执行 ON DUPLICATE KEY UPDATE 以获取现有记录的 ID 并将其与句子相关联。我希望能够直接在 ORM 中执行此操作，因为级联非常方便(上面的示例实际上是简化的......涉及到很多表)。由于 ORM 执行级联插入，我什至不知道如何控制字典中单词的插入。

这似乎是一个非常标准的问题，但我看到的大多数响应都不使用 ORM。我看过很多相关问题(关于重复 key 更新)，答案似乎是自定义的@compiles，例如:

https://github.com/bedwards/sqlalchemy_mysql_ext/blob/master/duplicate.py

我不明白的是(因为我还在学习 SQLAlchemy)这是否可以在 ORM 中使用？如果没有，是否有其他解决方案可以？

我已经尝试过使用 session.merge() 而不是 session.add() 但这没有用。

如有任何帮助，我们将不胜感激。

编辑:

感谢 iuridiniz 下面的帮助，我现在可以提供一些具体的代码来演示问题。假设我们有一个与用户具有一对多关系的组表，并且每个用户与(电子邮件)地址具有一对一关系，并且后者必须是唯一的。我们现在可以创建两个具有相同地址的用户，如果我们直接对用户进行合并，它的行为是正确的(不会因为重复的电子邮件地址而抛出错误)。但是，如果我们创建两个具有相同地址的用户，然后将他们添加到一个组并尝试在该组上进行合并，它将抛出一个唯一键错误(即它在尝试添加地址之前不检查地址是否存在):

from sqlalchemy import create_engine, Column, types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.orm import Session

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref


engine = create_engine('sqlite:///:memory:', echo=False)
Base = declarative_base()

session = scoped_session(sessionmaker(bind=engine))

class Group(Base):
    __tablename__ = "groups"
    gid = Column(types.Integer, primary_key=True)
    name = Column(types.String(255))

    users = relationship("User", backref="group")

    def __repr__(self):
        ret =  "Group(name=%r)" % self.name
        for user in self.users:
            ret += str(user)

class User(Base):
    __tablename__ = "users"
    login = Column(types.String(50), primary_key=True)
    name = Column(types.String(255))
    group_id = Column(types.Integer, ForeignKey('groups.gid'))
    address = Column(types.String(200), 
                            ForeignKey('addresses.email_address'))
    email = relationship("Address")

    def __repr__(self):
        return "User(login=%r, name=%r)\n%s" % (self.login, self.name,
                str(self.email))

有没有想过用优雅的方式解决这个问题？

对于最初使用“answer”发布此编辑的糟糕堆栈溢出礼仪，我们深表歉意。

Answer 1

AFAIK SQLAlchemy ORM 层无法执行ON DUPLICATE KEY UPDATE。幸运的是session.merge()函数可以为您执行此操作，但仅如果键是主键(您的情况是)。

session.merge(o) 通过发出 SELECT 检查是否存在具有相同主键值的行，如果为真，则发出 UPDATE 而不是 INSERT。

看这个例子:

from sqlalchemy import create_engine, Column, types
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

engine = create_engine('sqlite:///:memory:', echo=False)
Base = declarative_base()
session = scoped_session(sessionmaker(bind=engine))

class User(Base):
    __tablename__ = "user"
    login = Column(types.String(50), primary_key=True)
    name = Column(types.String(255))

    def __repr__(self):
        return "User(login=%r, name=%r)" % (self.login, self.name)

Base.metadata.create_all(engine)

if __name__ == '__main__':
    # create two users    
    u1 = User(login='iuridiniz', name="Iuri Diniz")
    u2 = User(login='someuser', name="Some User")
    session.merge(u1) # could be session.add(u1)
    session.merge(u2) # could be session.add(u2)
    session.commit()

    # print all users
    print("First two users")
    for u in session.query(User):
        print(u)

    # create more two users, one with the same login
    u3 = User(login='iuridiniz', name="Iuri Gomes Diniz")
    u4 = User(login='anotheruser', name="Another User")
    session.merge(u3) # session.add(u3) will raise a sqlalchemy.exc.IntegrityError
    session.merge(u4) # could be session.add(u4)
    session.commit()

    print("More two users")
    for u in session.query(User):
        print(u)

输出:

First two users
User(login=u'iuridiniz', name=u'Iuri Diniz')
User(login=u'someuser', name=u'Some User')
More two users
User(login=u'iuridiniz', name=u'Iuri Gomes Diniz')
User(login=u'someuser', name=u'Some User')
User(login=u'anotheruser', name=u'Another User')

将 engine = create_engine('sqlite:///:memory:', echo=False) 更改为 engine = create_engine('sqlite:///:memory:', echo =True) 为了查看执行的查询:

[INFO Engine] SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
[INFO Engine] ()
[INFO Engine] SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
[INFO Engine] ()
[INFO Engine] PRAGMA table_info("user")
[INFO Engine] ()
[INFO Engine] 
CREATE TABLE user (
    login VARCHAR(50) NOT NULL, 
    name VARCHAR(255), 
    PRIMARY KEY (login)
)


[INFO Engine] ()
[INFO Engine] COMMIT
[INFO Engine] BEGIN (implicit)
[INFO Engine] SELECT user.login AS user_login, user.name AS user_name 
FROM user 
WHERE user.login = ?
[INFO Engine] ('iuridiniz',)
[INFO Engine] INSERT INTO user (login, name) VALUES (?, ?)
[INFO Engine] ('iuridiniz', 'Iuri Diniz')
[INFO Engine] SELECT user.login AS user_login, user.name AS user_name 
FROM user 
WHERE user.login = ?
[INFO Engine] ('someuser',)
[INFO Engine] INSERT INTO user (login, name) VALUES (?, ?)
[INFO Engine] ('someuser', 'Some User')
[INFO Engine] COMMIT
[INFO Engine] BEGIN (implicit)
[INFO Engine] SELECT user.login AS user_login, user.name AS user_name 
FROM user 
WHERE user.login = ?
[INFO Engine] ('iuridiniz',)
[INFO Engine] UPDATE user SET name=? WHERE user.login = ?
[INFO Engine] ('Iuri Gomes Diniz', 'iuridiniz')
[INFO Engine] SELECT user.login AS user_login, user.name AS user_name 
FROM user 
WHERE user.login = ?
[INFO Engine] ('anotheruser',)
[INFO Engine] INSERT INTO user (login, name) VALUES (?, ?)
[INFO Engine] ('anotheruser', 'Another User')
[INFO Engine] COMMIT