我们知道MySQL支持insert update，其逻辑为：如果在INSERT后导致在一个UNIQUE索引或PRIMARY KEY中出现重复值，则执行UPDATE

>>> INSERT INTO test(data_id, device_id) VALUES (1, 2) ON DUPLICATE KEY UPDATE device_id = '2'; 

注意：与REPLACE INTO逻辑不一样，REPLACE INTO是在重复后，先将该条数据删除再进行插入，所以两者效果不同
参考：https://my.oschina.net/kingdelee/blog/388589

使用SQLAlchemy 实现与INSERT UPDATE相同的效果 (略微存在差异)，可以使用session.merge()

from sqlalchemy import MetaData
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session

metadata = MetaData()
metadata.reflect(engine, only=['test'])
Base = automap_base(metadata=metadata)
Base.prepare()
test = Base.classes.test
with Session(engine) as session, session.begin():
    try:
        # data_id和device_id中必须有一个为主键，没有主键而只有唯一索引会报错，这点与INSERT UPDATE不同
        session.merge(test(data_id = '1', device_id = '2'))
    except:
        session.rollback()
        raise
    else:
        session.commit()

参考：https://blog.csdn.net/u010339879/article/details/84529627

2021-12-27更新

仅支持MySQL

from sqlalchemy.dialects.mysql import insert
from sqlalchemy import Table

...

table = Table('iot_xj_liquid_fertilizer', metadata, autoload_with=engine)
insert_stmt = insert(table).values([{'data_id':'1'}, {'data_id':'2'}])
on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(data_id=insert_stmt.inserted.data_id)
result = engine.execute(on_duplicate_key_stmt)

如果想实现更新全部

from sqlalchemy.dialects.mysql import insert
from sqlalchemy import Table, text

...
data = [{'data_id':'1', 'device_id': '01'}, {'data_id':'2', 'device_id': '02'}]

table = Table('iot_xj_liquid_fertilizer', metadata, autoload_with=engine)
insert_stmt = insert(table).values(data)
kw={}
for key in data[0].keys():
    kw[key] = text(f'VALUES({key})')
on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(**kw)
result = engine.execute(on_duplicate_key_stmt)

参考：
https://stackoverflow.com/a/48373874/7151777
https://docs.sqlalchemy.org/en/14/dialects/mysql.html#insert-on-duplicate-key-update-upsert
https://www.cnblogs.com/better-farther-world2099/articles/11737376.html