使用Peewee EXCLUDED处理MySQL Aurora Upsert冲突的语法错误及竞态问题

问题背景 #

尝试使用Peewee执行Upsert操作，需求是仅当待插入数据的modifiedon（字符串格式时间戳，如2022-11-22T17:00:34.965Z）比数据库中现有数据新时，才更新条目。表的复合主键为(id, start_date, end_date)，当前代码使用Peewee的on_conflict方法，但执行后无效果，直接在MySQL中运行生成的SQL会触发语法错误。

原代码如下：

from peewee import EXCLUDED
query = (MyTable.insert(
    id=sched.id,
    idenc=sched.idenc,
    createdon=sched.createdon,
    modifiedon=sched.modifiedon,
    deletedon=sched.deletedon,
    canceledon=sched.canceledon,
    is_deleted=sched.is_deleted,
    start_date=sched.start_date,
    end_date=sched.end_date,
    label=sched.label
)
    .on_conflict(conflict_target=[MyTable.id,
                                  MyTable.start_date,
                                  MyTable.end_date],
                 update={
                     MyTable.modifiedon: EXCLUDED.modifiedon,
                     MyTable.label: EXCLUDED.label
    },
    where=(EXCLUDED.modifiedon > MyTable.modifiedon)))
query.execute()

生成的SQL语句：

(
    'INSERT INTO "mytable" ("id", "idenc", "createdon", "modifiedon", "deletedon", "canceledon", "isdeleted", "startdate", "enddate", "label") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT ("id", "startdate", "enddate") DO UPDATE SET "modifiedon" = EXCLUDED."modifiedon", "label" = EXCLUDED."label" WHERE
     (EXCLUDED."modifiedon" > "mytable"."modifiedon")',
    [ 'ymmqzHviWsMgabzPTEKU',
    'ebbe37ec-cb75-5bc6-9466-b170a458c469',
    '2022-11-22T17:00:05.175Z',
    '2022-12-05T11:23:31.563569Z',
    '',
    'None',
    False,
    '2022-11-21T18:01:00.000Z',
    '2022-12-31T17:59:00.000Z',
    '1' ]
)

此外，原方案使用get_or_create+replace存在竞态条件，需要更可靠的替代方案。

核心原因分析 #

1. 语法不兼容：Peewee的on_conflict方法生成的是PostgreSQL风格的ON CONFLICT语法，但你使用的是MySQL Aurora 5.7，MySQL并不支持该语法，它的Upsert实现是ON DUPLICATE KEY UPDATE，这是导致SQL语法错误的直接原因。
2. 字段类型隐患：modifiedon使用varchar存储时间戳，虽然ISO8601格式的字符串可以按字典序比较，但不如使用原生的datetime或timestamp类型可靠，容易出现格式不一致导致的比较错误。

正确实现方案 #

针对MySQL Aurora 5.7，使用Peewee的on_duplicate_key_update方法结合条件函数实现原子性Upsert，避免竞态条件：

1. 确保表定义正确 #

首先确认表的复合主键已正确定义：

from peewee import MySQLDatabase, Model, CharField, BooleanField, DateTimeField, CompositeKey, fn, EXCLUDED

db = MySQLDatabase('your_database', user='user', password='password', host='host')

class MyTable(Model):
    id = CharField()
    idenc = CharField()
    # 建议将时间字段改为DateTimeField，自动处理时间格式
    createdon = DateTimeField()
    modifiedon = DateTimeField()
    deletedon = DateTimeField(null=True)
    canceledon = DateTimeField(null=True)
    is_deleted = BooleanField()
    start_date = DateTimeField()
    end_date = DateTimeField()
    label = CharField()

    class Meta:
        database = db
        primary_key = CompositeKey('id', 'start_date', 'end_date')

2. 原子性Upsert代码 #

使用IF函数实现条件更新，仅当待插入数据的modifiedon更新时才覆盖字段：

query = (MyTable.insert(
    id=sched.id,
    idenc=sched.idenc,
    createdon=sched.createdon,
    modifiedon=sched.modifiedon,
    deletedon=sched.deletedon,
    canceledon=sched.canceledon,
    is_deleted=sched.is_deleted,
    start_date=sched.start_date,
    end_date=sched.end_date,
    label=sched.label
).on_duplicate_key_update(
    # 仅当EXCLUDED的modifiedon更大时，才更新modifiedon和label
    modifiedon=fn.IF(EXCLUDED.modifiedon > MyTable.modifiedon, EXCLUDED.modifiedon, MyTable.modifiedon),
    label=fn.IF(EXCLUDED.modifiedon > MyTable.modifiedon, EXCLUDED.label, MyTable.label)
))
query.execute()

3. 字段类型优化说明 #

如果坚持使用varchar存储时间戳，代码逻辑不变，但需确保所有时间字符串格式完全一致（包括时区、精度），避免比较错误。优先推荐使用DateTimeField，数据库会自动处理时间解析和比较，更可靠。

补充问题解答 #

1. 带WHERE条件的Upsert：MySQL的ON DUPLICATE KEY UPDATE不支持直接添加WHERE子句，但可以通过IF/CASE等条件函数在SET子句中实现逻辑判断，达到仅满足条件时更新的效果。
2. 非主键字段的Upsert：若要基于非主键字段触发Upsert，需先为该字段创建唯一索引，然后使用on_duplicate_key_update方法，Peewee完全支持这种场景。
3. 竞态条件解决：数据库级别的ON DUPLICATE KEY UPDATE是原子操作，整个插入/更新过程由数据库在单事务中完成，不会出现中间状态被其他进程修改的情况，彻底解决get_or_create方案的竞态问题。

内容的提问来源于stack exchange，提问作者Jimson James