如何解决Django中update_or_create引发的IntegrityError问题？

想请教为什么这里会抛出IntegrityError，update_or_create不是应该在对象存在时更新吗？

相关代码及错误信息如下：

models.py #

class OHLCV(TimestampedModel):
    market = models.ForeignKey(Market, on_delete=models.CASCADE, related_name='candles', null=True)
    index = models.ForeignKey(Index, on_delete=models.CASCADE, related_name='candles', null=True)
    timeframe = models.CharField(max_length=10)
    open, high, low, close, volume = [models.FloatField(null=True) for i in range(5)]
    datetime = models.DateTimeField(null=True)

    class Meta:
        verbose_name_plural = "OHLCV"
        unique_together = [['market', 'timeframe', 'datetime'],
                           ['index', 'timeframe', 'datetime']]

tasks.py #

@app.task
def bulk_update_ohlcv(timeframe):
    for obj in Market.objects.filter(active=True):
        if obj.exchange.is_status_ok():
            update_ohlcv.delay('market', obj.pk, timeframe)


@app.task
def update_ohlcv(self, type, pk, timeframe):

    [code here]
    if obj.__class__ == Market:
        ohlcv, created = OHLCV.objects.update_or_create(market=obj,
                                                        timeframe=timeframe,
                                                        datetime=dt,
                                                        open=candle[1],
                                                        defaults=defaults
                                                        )

    elif obj.__class__ == Index:
        ohlcv, created = OHLCV.objects.update_or_create(index=obj,
                                                        timeframe=timeframe,
                                                        datetime=dt,
                                                        open=candle[1],
                                                        defaults=defaults
                                                        )        

错误信息 #

-celery_worker-1  | 2023-04-01T06:55:47.710298043Z IntegrityError: duplicate key value violates unique constraint 
-celery_worker-1  | 2023-04-01T06:55:47.710302260Z "market_ohlcv_market_id_timeframe_datetime_8ffd84de_uniq"
-celery_worker-1  | 2023-04-01T06:55:47.710306227Z DETAIL:  Key (market_id, timeframe, datetime)=(84, 5m, 2023-03-31 21:20:00+00) 
-celery_worker-1  | 2023-04-01T06:55:47.710310646Z already exists.

原因分析 #

1. update_or_create参数误用：你把open字段加入了查询条件，但数据库的唯一约束是(market, timeframe, datetime)。当数据库中已有一条唯一键匹配但open值不同的记录时，update_or_create会因查询不到完全匹配的记录而尝试创建新条目，直接触发唯一约束冲突。
2. Celery并发竞态：多个Celery任务同时处理同一条OHLCV记录时，可能出现多个任务都检测到记录“不存在”，同时发起创建请求的情况，进而触发唯一约束错误。

解决方案 #

1. 修正update_or_create查询逻辑 #

只保留唯一约束中的字段作为查询条件，将open移到defaults字典中，让方法基于唯一键匹配记录：

# 处理Market类型
ohlcv, created = OHLCV.objects.update_or_create(
    market=obj,
    timeframe=timeframe,
    datetime=dt,
    defaults={**defaults, 'open': candle[1]}
)

# 处理Index类型同理
ohlcv, created = OHLCV.objects.update_or_create(
    index=obj,
    timeframe=timeframe,
    datetime=dt,
    defaults={**defaults, 'open': candle[1]}
)

2. 解决并发竞态问题 #

• 事务+行锁：用事务包裹操作，同时通过select_for_update锁定目标记录，防止多任务同时修改：

  from django.db import transaction
  
  with transaction.atomic():
      # 锁定匹配的记录（如果存在）
      qs = OHLCV.objects.filter(market=obj, timeframe=timeframe, datetime=dt).select_for_update()
      if qs.exists():
          qs.update(**defaults, open=candle[1])
          ohlcv = qs.first()
          created = False
      else:
          ohlcv = OHLCV.objects.create(market=obj, timeframe=timeframe, datetime=dt, open=candle[1], **defaults)
          created = True
  

• 任务去重：给Celery任务添加去重机制，确保同一个(market, timeframe, datetime)组合不会被多个任务同时处理。

3. 清理脏数据（可选） #

如果数据库中存在违反唯一约束的重复记录，需要先清理这些数据，确保约束的有效性。

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