SQLAlchemy ON DUPLICATE KEY UPDATE

Question

INSERT ... ON DUPLICATE KEY UPDATE在SQLAlchemy中有一种优雅的方式吗？我的意思是语法类似的东西inserter.insert().execute(list_of_dictionaries)？

Answer 1

ON DUPLICATE KEY UPDATE 发布版本为1.2的MySQL

此功能现在内置于SQLAlchemy for MySQL中.somada141的答案如下是最佳解决方案:https://stackoverflow.com/a/48373874/319066

ON DUPLICATE KEY UPDATE 在SQL语句中

如果您希望生成的SQL实际包含ON DUPLICATE KEY UPDATE,最简单的方法是使用@compiles装饰器.

可以在github上找到代码(从reddit主题上的一个好主题链接)作为示例:

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert)
def append_string(insert, compiler, **kw):
    s = compiler.visit_insert(insert, **kw)
    if 'append_string' in insert.kwargs:
        return s + " " + insert.kwargs['append_string']
    return s


my_connection.execute(my_table.insert(append_string = 'ON DUPLICATE KEY UPDATE foo=foo'), my_values)

但请注意,在此方法中,您必须手动创建append_string.您可以更改append_string函数,以便它自动将插入字符串更改为带有"ON DUPLICATE KEY UPDATE"字符串的插入,但由于懒惰,我不打算这样做.

ON DUPLICATE KEY UPDATE ORM中的功能

SQLAlchemy的不提供到接口ON DUPLICATE KEY UPDATE或MERGE或在其ORM层的任何其他类似的功能.然而,它具有只有在所讨论的密钥是主键时session.merge()才能复制功能的功能.

session.merge(ModelObject)首先通过发送SELECT查询(或通过在本地查找)来检查是否存在具有相同主键值的行.如果是这样,它会在某处设置一个标志,指示ModelObject已经在数据库中,并且SQLAlchemy应该使用UPDATE查询.请注意,merge比这复杂得多,但它可以很好地复制主键的功能.

但是,如果您想要ON DUPLICATE KEY UPDATE使用非主键(例如,另一个唯一键)的功能,该怎么办？不幸的是,SQLAlchemy没有任何这样的功能.相反,你必须创建类似Django的东西get_or_create().另一个StackOverflow答案涵盖了它,为了方便起见,我将在这里粘贴一个经过修改的工作版本.

def get_or_create(session, model, defaults=None, **kwargs):
    instance = session.query(model).filter_by(**kwargs).first()
    if instance:
        return instance
    else:
        params = dict((k, v) for k, v in kwargs.iteritems() if not isinstance(v, ClauseElement))
        if defaults:
            params.update(defaults)
        instance = model(**params)
        return instance

Answer 2

我应该提一下,自从v1.2发布以来,SQLAlchemy'核心'有一个内置的解决方案,内置并可以在这里看到(下面复制的片段):

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='inserted value')

on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
    data=insert_stmt.inserted.data,
    status='U'
)

conn.execute(on_duplicate_key_stmt)