Django调用update_or_create触发PostgreSQL主键约束错误如何解决？

问题根源 #

1. 你错误使用了update_or_create的参数结构。该方法参数分为两部分：第一部分为查询匹配条件，用于判断数据库中是否已有对应记录；第二部分为defaults字典，存放匹配到记录时要更新、未匹配到时要新增的字段值。你当前将所有字段都放在查询条件位置，完全不符合方法设计逻辑。
2. 报错提示的是YouTubeCurrentChannelStat的主键冲突，但你给出的代码仅操作了YouTubeVideoItemsCurrentStat，说明你在调用update_videos_db_records的上游逻辑中，尝试重复创建同一个channel_id的频道记录，这是冲突的直接来源。

修复方案 #

1. 修正update_or_create写法

YouTubeVideoItemsCurrentStat的主键是video_id，直接用该字段作为唯一匹配条件，其他待更新字段全部放入defaults即可：

def update_videos_db_records(channel_id, response, video_id):
    YouTubeVideoItemsCurrentStat.objects.update_or_create(
        video_id=video_id,
        defaults={
            "channel_id": channel_id,
            "stat_data": {
                "title": response.get('title', ''),
                "likes": response.get('likes', 0),
                "views": response.get('views', 0),
                "thumbnail": response.get('thumb_url', ''),
                "comments": response.get('comments', 0),
                "dislikes": response.get('dislikes', 0),
                "favourites": response.get('favourite', 0)
            },
            "last_updated": datetime.datetime.now()
        }
    )

2. 修复上游频道创建逻辑

频道表主键冲突的问题，同样用update_or_create处理写入逻辑即可，示例如下：

# 可根据你的业务字段调整参数
YouTubeCurrentChannelStat.objects.update_or_create(
    channel_id=你的频道ID变量,
    defaults={
        "blogger": 关联的BloggerProfile实例,
        "channel_title": 频道名称,
        "subscribers": 订阅数,
        "views": 总播放量,
        # 其他需要更新的频道字段
        "last_updated": datetime.datetime.now()
    }
)

3. 并发场景兼容（可选）

如果你的更新任务是Celery多worker并发执行，可能出现同一时间多个任务尝试创建同一条记录的情况，可以加事务和异常捕获兼容：

from django.db import IntegrityError, transaction

def update_videos_db_records(channel_id, response, video_id):
    try:
        with transaction.atomic():
            YouTubeVideoItemsCurrentStat.objects.update_or_create(
                video_id=video_id,
                defaults={
                    "channel_id": channel_id,
                    "stat_data": {
                        "title": response.get('title', ''),
                        "likes": response.get('likes', 0),
                        "views": response.get('views', 0),
                        "thumbnail": response.get('thumb_url', ''),
                        "comments": response.get('comments', 0),
                        "dislikes": response.get('dislikes', 0),
                        "favourites": response.get('favourite', 0)
                    },
                    "last_updated": datetime.datetime.now()
                }
            )
    except IntegrityError:
        # 捕获冲突后直接走更新逻辑即可
        YouTubeVideoItemsCurrentStat.objects.filter(video_id=video_id).update(
            channel_id=channel_id,
            stat_data={
                "title": response.get('title', ''),
                "likes": response.get('likes', 0),
                "views": response.get('views', 0),
                "thumbnail": response.get('thumb_url', ''),
                "comments": response.get('comments', 0),
                "dislikes": response.get('dislikes', 0),
                "favourites": response.get('favourite', 0)
            },
            last_updated=datetime.datetime.now()
        )

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