Django添加唯一约束时,如何跳过现有重复数据避免完整性错误?
问题
我定义了如下AdvisorResponse模型,想要实现一位顾问仅能拥有一条状态为Accepted的响应,因此在Meta类中添加了条件唯一约束:
class AdvisorResponse(UUIDTimeStampedModel): chat_request = models.ForeignKey(ChatRequest, on_delete=models.CASCADE, related_name='advisor_responses', null=True) advisor = models.ForeignKey(User, on_delete=models.CASCADE, null=True, blank=True) status = models.CharField(max_length=10, choices=ADVISOR_REQUEST_STATUS, default=ADVISOR_REQUEST_STATUS_ACCEPTED) # Data field holds calculated indexed data which is for UI consumption and doesn't change over time. data = JSONField(default=dict, blank=True) class Meta: constraints = [ models.UniqueConstraint( fields=['advisor', 'status'], name='unique_advisor_accepted_response', condition=models.Q(status=ADVISOR_REQUEST_STATUS_ACCEPTED) ) ]
但执行迁移时,因为现有数据库中存在同一顾问多条Accepted状态的响应,触发了完整性错误:
psycopg2.errors.UniqueViolation: could not create unique index "unique_advisor_accepted_response" DETAIL: Key (advisor_id, status)=(11, accepted) is duplicated
我不想删除生产环境的数据,请问如何执行迁移,让约束仅作用于新插入的数据,跳过原有冲突数据?
解决方案
要实现约束仅对新数据生效,同时保留现有历史数据,分三步操作:
1. 创建不验证现有数据的唯一约束
运行makemigrations生成迁移文件后,打开对应的迁移文件,修改约束创建逻辑,添加concurrently=True和not_valid=True参数(仅适用于PostgreSQL):
from django.db import migrations, models class Migration(migrations.Migration): dependencies = [ # 保留原有的依赖迁移配置 ] operations = [ migrations.AddConstraint( model_name='advisorresponse', constraint=models.UniqueConstraint( fields=['advisor', 'status'], name='unique_advisor_accepted_response', condition=models.Q(status=ADVISOR_REQUEST_STATUS_ACCEPTED) ), # 添加这两个参数跳过现有数据校验 concurrently=True, not_valid=True, ), ]
注意:使用
CONCURRENTLY时,需要带--no-transaction参数运行迁移:python manage.py migrate --no-transaction
2. 处理现有冲突数据(可选但建议)
不删除数据的前提下,可将冲突的历史响应标记为其他状态(比如archived),避免后续潜在问题:
# 在Django Shell中执行,或编写数据迁移脚本 from your_app.models import AdvisorResponse, ADVISOR_REQUEST_STATUS_ACCEPTED, ADVISOR_REQUEST_STATUS_ARCHIVED # 遍历所有有重复Accepted响应的顾问 for advisor in User.objects.filter(advisorresponse__status=ADVISOR_REQUEST_STATUS_ACCEPTED).distinct(): # 按创建时间倒序取该顾问的Accepted响应 responses = AdvisorResponse.objects.filter(advisor=advisor, status=ADVISOR_REQUEST_STATUS_ACCEPTED).order_by('-created_at') # 保留最新的一条,其余改为归档状态 for response in responses[1:]: response.status = ADVISOR_REQUEST_STATUS_ARCHIVED response.save()
3. 验证现有数据(可选)
后续清理完所有冲突数据后,可执行SQL让数据库验证现有数据是否符合约束:
ALTER TABLE your_app_advisorresponse VALIDATE CONSTRAINT unique_advisor_accepted_response;
也可以通过Django迁移文件的RunSQL操作执行这条语句。
内容的提问来源于stack exchange,提问作者GuruRandapa
相关产品推荐
相关产品推荐

